Fix #813

Signed-off-by: Francesco Guardiani <francescoguard@gmail.com>
knative-extensions · Apr 16, 2021 · 84baddd · 84baddd
1 parent 4c7f76d
commit 84baddd
Show file tree

Hide file tree

Showing 2 changed files with 11 additions and 1 deletion.
diff --git a/...va/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManager.java b/...va/dev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManager.java
@@ -251,7 +251,7 @@ private int blockIndex(long val) {
     }
 
     private void checkAcksArraySize(int blockIndex) {
-      if (this.uncommitted.length < blockIndex) {
+      if (blockIndex > this.uncommitted.length - 1) {
         // Let's make sure we create enough room for more unordered records
         this.uncommitted = Arrays.copyOf(this.uncommitted, (blockIndex + 1) * 2);
       }

diff --git a/...ev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManagerTest.java b/...ev/knative/eventing/kafka/broker/dispatcher/consumer/impl/UnorderedOffsetManagerTest.java
@@ -50,6 +50,16 @@ public void shouldCommitAfterSendingEventsOrderedOnTheSamePartition() {
       .containsEntry(new TopicPartition("aaa", 0), 10L);
   }
 
+  @Test
+  public void shouldNotCommitAndNotGoOutOfBounds() {
+    assertThatOffsetCommitted(List.of(new TopicPartition("aaa", 0)), offsetStrategy -> {
+      offsetStrategy.recordReceived(record("aaa", 0, 0));
+      offsetStrategy.successfullySentToSubscriber(record("aaa", 0, 64));
+      offsetStrategy.successfullySentToSubscriber(record("aaa", 0, 128));
+    })
+      .isEmpty();
+  }
+
   @Test
   public void shouldCommitAfterSendingEventsOrderedOnTheSamePartitionWithInducedFailure() {
     assertThatOffsetCommittedWithFailures(List.of(new TopicPartition("aaa", 0)), (offsetStrategy, failureFlag) -> {