fix(mock consumer): HighWaterMarkOffset

IBM · Mar 3, 2023 · aa0f7ee · aa0f7ee
1 parent 397cee4
commit aa0f7ee
Show file tree

Hide file tree

Showing 2 changed files with 6 additions and 1 deletion.
diff --git a/mocks/consumer.go b/mocks/consumer.go
@@ -343,7 +343,7 @@ func (pc *PartitionConsumer) Messages() <-chan *sarama.ConsumerMessage {
 }
 
 func (pc *PartitionConsumer) HighWaterMarkOffset() int64 {
-	return atomic.LoadInt64(&pc.highWaterMarkOffset) + 1
+	return atomic.LoadInt64(&pc.highWaterMarkOffset)
 }
 
 // Pause implements the Pause method from the sarama.PartitionConsumer interface.

diff --git a/mocks/consumer_test.go b/mocks/consumer_test.go
@@ -395,6 +395,11 @@ func TestConsumerOffsetsAreManagedCorrectlyWithSpecifiedOffset(t *testing.T) {
 	if len(trm.errors) != 0 {
 		t.Errorf("Expected to not report any errors, found: %v", trm.errors)
 	}
+
+	if pc.HighWaterMarkOffset() != message2.Offset+1 {
+		diff := pc.HighWaterMarkOffset() - message2.Offset
+		t.Errorf("Difference between highwatermarkoffset and last message offset greater than 1, got: %v", diff)
+	}
 }
 
 func TestConsumerInvalidConfiguration(t *testing.T) {