Skip to content

Commit

Permalink
Merge pull request #2082 from grongor/fix-high-watermark-partition-co…
Browse files Browse the repository at this point in the history
…nsumer

fix: set HWMO during creation of partitionConsumer (fix incorrect HWMO before first fetch)
  • Loading branch information
dnwe authored Dec 7, 2021
2 parents 75762af + 179dc9d commit 024e359
Show file tree
Hide file tree
Showing 2 changed files with 12 additions and 0 deletions.
3 changes: 3 additions & 0 deletions consumer.go
Original file line number Diff line number Diff line change
Expand Up @@ -400,6 +400,9 @@ func (child *partitionConsumer) chooseStartingOffset(offset int64) error {
if err != nil {
return err
}

child.highWaterMarkOffset = newestOffset

oldestOffset, err := child.consumer.client.GetOffset(child.topic, child.partition, OffsetOldest)
if err != nil {
return err
Expand Down
9 changes: 9 additions & 0 deletions consumer_test.go
Original file line number Diff line number Diff line change
Expand Up @@ -57,6 +57,9 @@ func TestConsumerOffsetManual(t *testing.T) {
}

// Then
if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewest {
t.Errorf("Expected high water mark offset %d, found %d", offsetNewest, hwmo)
}
for i := int64(0); i < 10; i++ {
select {
case message := <-consumer.Messages():
Expand Down Expand Up @@ -109,6 +112,9 @@ func TestConsumerOffsetNewest(t *testing.T) {
}

// Then
if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewest {
t.Errorf("Expected high water mark offset %d, found %d", offsetNewest, hwmo)
}
assertMessageOffset(t, <-consumer.Messages(), 10)
if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewestAfterFetchRequest {
t.Errorf("Expected high water mark offset %d, found %d", offsetNewestAfterFetchRequest, hwmo)
Expand Down Expand Up @@ -154,6 +160,9 @@ func TestConsumerOffsetOldest(t *testing.T) {
}

// Then
if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewest {
t.Errorf("Expected high water mark offset %d, found %d", offsetNewest, hwmo)
}
assertMessageOffset(t, <-consumer.Messages(), int64(7))
if hwmo := consumer.HighWaterMarkOffset(); hwmo != offsetNewest {
t.Errorf("Expected high water mark offset %d, found %d", offsetNewest, hwmo)
Expand Down

0 comments on commit 024e359

Please sign in to comment.