pass one offset at a time

Signed-off-by: Yashash H L <[email protected]>

pkg/apis/proto/source/v1/source.proto

@@ -78,11 +78,8 @@ message ReadResponse {
  */
 message AckRequest {
   message Request {
-    // Required field holding a list of offsets to be acknowledged.
-    // The offsets must be strictly corresponding to the previously read batch,
-    // meaning the offsets must be in the same order as the datum responses in the ReadResponse.
-    // By enforcing ordering, we can save deserialization effort on the server side, assuming the server keeps a local copy of the raw/un-serialized offsets.
-    repeated Offset offsets = 1;
+    // Required field holding the offset to be acked
+    Offset offset = 1;
   }
   // Required field holding the request. The list will be ordered and will have the same order as the original Read response.
   Request request = 1;

pkg/sourcer/service.go

@@ -138,12 +138,10 @@ func (fs *Service) AckFn(stream sourcepb.Source_AckFnServer) error {
 			return err
 		}
 
-		for _, offset := range req.Request.GetOffsets() {
-			request := ackRequest{
-				offset: NewOffset(offset.GetOffset(), offset.GetPartitionId()),
-			}
-			fs.Source.Ack(ctx, &request)
+		request := ackRequest{
+			offset: NewOffset(req.Request.Offset.GetOffset(), req.Request.Offset.GetPartitionId()),
 		}
+		fs.Source.Ack(ctx, &request)
 	}
 }
 

pkg/sourcer/service_test.go

@@ -131,7 +131,7 @@ func (a *AckFnServerTest) Recv() (*sourcepb.AckRequest, error) {
 	a.index++
 	return &sourcepb.AckRequest{
 		Request: &sourcepb.AckRequest_Request{
-			Offsets: []*sourcepb.Offset{offset},
+			Offset: offset,
 		},
 	}, nil
 }
@@ -146,7 +146,7 @@ func NewAckFnServerTest(
 	}
 }
 
-func (a *AckFnServerTest) SendAndClose(response *sourcepb.AckResponse) error {
+func (a *AckFnServerTest) SendAndClose(*sourcepb.AckResponse) error {
 	return nil
 }