update example

Signed-off-by: Yashash H L <[email protected]>
numaproj · Oct 30, 2024 · dfa1117 · dfa1117
1 parent f681938
commit dfa1117
Showing 1 changed file with 3 additions and 2 deletions.
diff --git a/pkg/sourcer/examples/simple_source/impl/simple_source.go b/pkg/sourcer/examples/simple_source/impl/simple_source.go
@@ -76,8 +76,9 @@ func (s *SimpleSource) Read(_ context.Context, readRequest sourcesdk.ReadRequest
 func (s *SimpleSource) Ack(_ context.Context, request sourcesdk.AckRequest) {
 	s.lock.Lock()
 	defer s.lock.Unlock()
-	offset := deserializeOffset(request.Offset().Value())
-	delete(s.toAckSet, offset)
+	for _, offset := range request.Offsets() {
+		delete(s.toAckSet, deserializeOffset(offset.Value()))
+	}
 }
 
 func (s *SimpleSource) Partitions(_ context.Context) []int32 {