Update to AssignOffsets function to avoid purge of current assignment (…

…#74)

* Addition of support for AssignOffsets function to augment or modify a current list of assignments

* Update to add additional checking for calls to AssignOffsets with items that already exist or do not

* update to reflect requested change for removing current assignment

Co-authored-by: Conor McCarthy <conormccarthy@brainpool1.mynet>
Co-authored-by: mshimizu-kx <40049399+mshimizu-kx@users.noreply.github.com>
Co-authored-by: Conor McCarthy <conormccarthy@brainpool1.local>

kfk.c

@@ -549,7 +549,7 @@ EXP K1(kfkUnsub){
 }
 
 // https://github.com/edenhill/librdkafka/wiki/Manually-setting-the-consumer-start-offset
-EXP K3(kfkAssignOffsets){
+EXP K3(kfkassignOffsets){
   rd_kafka_t *rk;
   rd_kafka_topic_partition_list_t *t_partition;
   rd_kafka_resp_err_t err;
@@ -559,7 +559,9 @@ EXP K3(kfkAssignOffsets){
     return KNL;
   if(!(rk= clientIndex(x)))
     return KNL;
-  t_partition = rd_kafka_topic_partition_list_new(z->n);
+  // retrieve the current assignment
+  if(KFK_OK != (err=rd_kafka_assignment(rk, &t_partition)))
+    return krr((S)rd_kafka_err2str(err));
   plistoffsetdict(y->s,z,t_partition);
   if(KFK_OK != (err=rd_kafka_assign(rk,t_partition)))
     return krr((S) rd_kafka_err2str(err));

kfk.q

@@ -49,8 +49,8 @@ funcs:(
 	(`kfkPositionOffsets;3);
 	  // .kfk.CommittedOffsets[client_id:i;topic:s;partition_offsets:I!J]:partition_offsets
 	(`kfkCommittedOffsets;3);
-	  // .kfk.AssignOffsets[client_id:i;topic:s;partition_offsets:I!J]:()
-	(`kfkAssignOffsets;3);
+	  // .kfk.assignOffsets[client_id:i;topic:s;partition_offsets:I!J]:()
+	(`kfkassignOffsets;3);
           // .kfk.Threadcount[]:i
         (`kfkThreadCount;1);
           // .kfk.VersionSym[]:s
@@ -185,6 +185,22 @@ ClientMemberId:{[cid]
 
 // Assignment API logic
 
+// Run assignment of a topic to a particular partition making
+/* cid      = Integer denoting the client ID
+/* top      = Topic to be subscribed to as a symbol
+/* partoff  = Dictionary mapping integer partition to long offset location
+AssignOffsets:{[cid;top;partoff]
+  toppar:(count[partoff]#top)!key partoff;
+  tplist:distinct(,'/)(key;{"j"$value x})@\:toppar;
+  // Find locations where the current assigment needs to be overwritten
+  loc:where i.compAssign[cid;tplist];
+  if[count loc;
+    currentAssign:(!). flip tplist loc;
+    AssignDel[cid;currentAssign]
+    ];
+  assignOffsets[cid;top;partoff]
+  }
+
 // Assign a new topic-partition dictionary to be consumed by a designated clientid
 /* cid    = Integer denoting client ID
 /* toppar = Symbol!Long dictionary mapping the name of a topic to an associated partition