content: /*
   * Licensed to the Apache Software Foundation (ASF) under one or more
   * contributor license agreements. See the NOTICE file distributed with
   * this work for additional information regarding copyright ownership.
   * The ASF licenses this file to You under the Apache License, Version 2.0
   * (the "License"); you may not use this file except in compliance with
   * the License. You may obtain a copy of the License at
   *
   *    http://www.apache.org/licenses/LICENSE-2.0
   *
   * Unless required by applicable law or agreed to in writing, software
   * distributed under the License is distributed on an "AS IS" BASIS,
   * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
   * See the License for the specific language governing permissions and
   * limitations under the License.
   */
  package org.apache.kafka.clients.consumer.internals;
  
  import org.apache.kafka.clients.consumer.AcknowledgeType;
  
  import org.junit.jupiter.api.Test;
  
  import java.util.List;
  
  import static org.junit.jupiter.api.Assertions.assertEquals;
  import static org.junit.jupiter.api.Assertions.assertNotEquals;
  import static org.junit.jupiter.api.Assertions.assertTrue;
  
  public class AcknowledgementsTest {
  
      private final Acknowledgements acks = Acknowledgements.empty();
      private final int maxRecordsWithSameAcknowledgeType = Acknowledgements.MAX_RECORDS_WITH_SAME_ACKNOWLEDGE_TYPE;
  
      @Test
      public void testEmptyBatch() {
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertTrue(ackList.isEmpty());
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertTrue(ackList2.isEmpty());
      }
  
      @Test
      public void testSingleStateSingleRecord() {
          acks.add(0L, AcknowledgeType.ACCEPT);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(0L, ackList.get(0).lastOffset());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(0L, ackList2.get(0).lastOffset());
          assertEquals(1, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
      }
  
      @Test
      public void testSingleStateMultiRecord() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.ACCEPT);
          acks.add(2L, AcknowledgeType.ACCEPT);
          acks.add(3L, AcknowledgeType.ACCEPT);
          acks.add(4L, AcknowledgeType.ACCEPT);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(4L, ackList.get(0).lastOffset());
          assertNotEquals(0, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(4L, ackList2.get(0).lastOffset());
          assertNotEquals(0, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
      }
  
      @Test
      public void testSingleAcknowledgementTypeExceedingLimit() {
          int i = 0;
          for (; i < maxRecordsWithSameAcknowledgeType; i++) {
              acks.add(i, AcknowledgeType.ACCEPT);
          }
          acks.add(i++, AcknowledgeType.ACCEPT);
          acks.add(i++, AcknowledgeType.ACCEPT);
          for (int j = 0; j <= maxRecordsWithSameAcknowledgeType; j++) {
              acks.add(i + j, AcknowledgeType.REJECT);
          }
  
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(2, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 1, ackList.get(0).lastOffset());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(maxRecordsWithSameAcknowledgeType + 2, ackList.get(1).firstOffset());
          assertEquals(i + maxRecordsWithSameAcknowledgeType, ackList.get(1).lastOffset());
          assertEquals(1, ackList.get(1).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.REJECT.id, ackList.get(1).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(2, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 1, ackList2.get(0).lastOffset());
          assertEquals(1, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(maxRecordsWithSameAcknowledgeType + 2, ackList2.get(1).firstOffset());
          assertEquals(i + maxRecordsWithSameAcknowledgeType, ackList2.get(1).lastOffset());
          assertEquals(1, ackList2.get(1).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.REJECT.id, ackList2.get(1).acknowledgeTypes().get(0));
      }
  
      @Test
      public void testSingleAcknowledgementTypeWithGap() {
          for (int i = 0; i < maxRecordsWithSameAcknowledgeType; i++) {
              acks.add(i, null);
          }
          acks.add(maxRecordsWithSameAcknowledgeType, null);
          acks.add(maxRecordsWithSameAcknowledgeType + 1, null);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 1, ackList.get(0).lastOffset());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 1, ackList2.get(0).lastOffset());
          assertEquals(1, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(0));
      }
  
      @Test
      public void testOptimiseBatches() {
          int offset = 0;
          for (; offset < maxRecordsWithSameAcknowledgeType; offset++) {
              acks.add(offset, AcknowledgeType.ACCEPT);
          }
          acks.add(offset++, AcknowledgeType.REJECT);
          acks.add(offset++, AcknowledgeType.ACCEPT);
          acks.add(offset++, AcknowledgeType.RELEASE);
          acks.add(offset++, null);
  
          // Adding more than the max records
          for (int j = 0; j <= maxRecordsWithSameAcknowledgeType; j++) {
              acks.add(offset + j, AcknowledgeType.ACCEPT);
          }
          offset += maxRecordsWithSameAcknowledgeType + 1;
  
          // Adding 2 more records of different type
          acks.add(offset++, AcknowledgeType.REJECT);
          acks.add(offset, AcknowledgeType.RELEASE);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(3, ackList.size());
          assertEquals(0, ackList.get(0).firstOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 3, ackList.get(0).lastOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 4, ackList.get(1).firstOffset());
          assertEquals(2 * maxRecordsWithSameAcknowledgeType + 4, ackList.get(1).lastOffset());
          assertEquals(1, ackList.get(1).acknowledgeTypes().size());
          assertEquals(offset - 1, ackList.get(2).firstOffset());
          assertEquals(offset, ackList.get(2).lastOffset());
          assertEquals(2, ackList.get(2).acknowledgeTypes().size());
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(3, ackList2.size());
          assertEquals(0, ackList2.get(0).firstOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 3, ackList2.get(0).lastOffset());
          assertEquals(maxRecordsWithSameAcknowledgeType + 4, ackList2.get(1).firstOffset());
          assertEquals(2 * maxRecordsWithSameAcknowledgeType + 4, ackList2.get(1).lastOffset());
          assertEquals(1, ackList2.get(1).acknowledgeTypes().size());
          assertEquals(offset - 1, ackList2.get(2).firstOffset());
          assertEquals(offset, ackList2.get(2).lastOffset());
          assertEquals(2, ackList2.get(2).acknowledgeTypes().size());
      }
  
      @Test
      public void testSingleAcknowledgementTypeWithinLimit() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.ACCEPT);
          acks.add(2L, AcknowledgeType.ACCEPT);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
      }
  
      @Test
      public void testMultiStateMultiRecord() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.ACCEPT);
          acks.add(2L, AcknowledgeType.ACCEPT);
          acks.add(3L, AcknowledgeType.RELEASE);
          acks.add(4L, AcknowledgeType.RELEASE);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(4L, ackList.get(0).lastOffset());
          assertEquals(5, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(1));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(4));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(4L, ackList2.get(0).lastOffset());
          assertEquals(5, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(1));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(4));
      }
  
      @Test
      public void testMultiStateSingleMultiRecord() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.RELEASE);
          acks.add(2L, AcknowledgeType.RELEASE);
          acks.add(3L, AcknowledgeType.RELEASE);
          acks.add(4L, AcknowledgeType.RELEASE);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(4L, ackList.get(0).lastOffset());
          assertEquals(5, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(1));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(4));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(4L, ackList2.get(0).lastOffset());
          assertEquals(5, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(1));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(4));
      }
  
      @Test
      public void testMultiStateMultiSingleRecord() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.ACCEPT);
          acks.add(2L, AcknowledgeType.ACCEPT);
          acks.add(3L, AcknowledgeType.ACCEPT);
          acks.add(4L, AcknowledgeType.RELEASE);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(4L, ackList.get(0).lastOffset());
          assertEquals(5, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(1));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(4));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(4L, ackList2.get(0).lastOffset());
          assertEquals(5, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(1));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(4));
      }
  
      @Test
      public void testSingleGap() {
          acks.addGap(0L);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(0L, ackList.get(0).lastOffset());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(0L, ackList2.get(0).lastOffset());
          assertEquals(1, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(0));
      }
  
      @Test
      public void testMultiGap() {
          acks.addGap(0L);
          acks.addGap(1L);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(1L, ackList.get(0).lastOffset());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(1L, ackList2.get(0).lastOffset());
          assertEquals(1, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(0));
      }
  
      @Test
      public void testSingleGapSingleState() {
          acks.addGap(0L);
          acks.add(1L, AcknowledgeType.ACCEPT);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(1L, ackList.get(0).lastOffset());
          assertEquals(2, ackList.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(1));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(1L, ackList2.get(0).lastOffset());
          assertEquals(2, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(1));
      }
  
      @Test
      public void testSingleStateSingleGap() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.addGap(1L);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(1L, ackList.get(0).lastOffset());
          assertEquals(2, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(1));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(1L, ackList2.get(0).lastOffset());
          assertEquals(2, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(1));
      }
  
      @Test
      public void testMultiStateMultiGap() {
          acks.add(0L, AcknowledgeType.RELEASE);
          acks.addGap(1L);
          acks.addGap(2L);
          acks.add(3L, AcknowledgeType.ACCEPT);
          acks.add(4L, AcknowledgeType.ACCEPT);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(4L, ackList.get(0).lastOffset());
          assertEquals(5, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(1));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(4));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(4L, ackList2.get(0).lastOffset());
          assertEquals(5, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(1));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(3));
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(4));
      }
  
      @Test
      public void testMultiStateMultiGaps() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.RELEASE);
          acks.addGap(2L);
          acks.add(3L, AcknowledgeType.RELEASE);
          acks.addGap(4L);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(1, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(4L, ackList.get(0).lastOffset());
          assertEquals(5, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(1));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(3));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(4));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(1, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(4L, ackList2.get(0).lastOffset());
          assertEquals(5, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(1));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(2));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(3));
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(4));
      }
  
      @Test
      public void testNoncontiguousBatches() {
          acks.add(0L, AcknowledgeType.ACCEPT);
          acks.add(1L, AcknowledgeType.RELEASE);
          acks.add(3L, AcknowledgeType.REJECT);
          acks.add(4L, AcknowledgeType.REJECT);
          acks.add(6L, AcknowledgeType.REJECT);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(3, ackList.size());
          assertEquals(0L, ackList.get(0).firstOffset());
          assertEquals(1L, ackList.get(0).lastOffset());
          assertEquals(2, ackList.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.RELEASE.id, ackList.get(0).acknowledgeTypes().get(1));
          assertEquals(3L, ackList.get(1).firstOffset());
          assertEquals(4L, ackList.get(1).lastOffset());
          assertEquals(1, ackList.get(1).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.REJECT.id, ackList.get(1).acknowledgeTypes().get(0));
          assertEquals(6L, ackList.get(2).firstOffset());
          assertEquals(6L, ackList.get(2).lastOffset());
          assertEquals(1, ackList.get(2).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.REJECT.id, ackList.get(2).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(3, ackList2.size());
          assertEquals(0L, ackList2.get(0).firstOffset());
          assertEquals(1L, ackList2.get(0).lastOffset());
          assertEquals(2, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.ACCEPT.id, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(AcknowledgeType.RELEASE.id, ackList2.get(0).acknowledgeTypes().get(1));
          assertEquals(3L, ackList2.get(1).firstOffset());
          assertEquals(4L, ackList2.get(1).lastOffset());
          assertEquals(1, ackList2.get(1).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.REJECT.id, ackList2.get(1).acknowledgeTypes().get(0));
          assertEquals(6L, ackList2.get(2).firstOffset());
          assertEquals(6L, ackList2.get(2).lastOffset());
          assertEquals(1, ackList2.get(2).acknowledgeTypes().size());
          assertEquals(AcknowledgeType.REJECT.id, ackList2.get(2).acknowledgeTypes().get(0));
      }
  
  
      @Test
      public void testNoncontiguousGaps() {
          acks.addGap(2L);
          acks.addGap(4L);
  
          List<AcknowledgementBatch> ackList = acks.getAcknowledgementBatches();
          assertEquals(2, ackList.size());
          assertEquals(2L, ackList.get(0).firstOffset());
          assertEquals(2L, ackList.get(0).lastOffset());
          assertEquals(1, ackList.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(0).acknowledgeTypes().get(0));
          assertEquals(4L, ackList.get(1).firstOffset());
          assertEquals(4L, ackList.get(1).lastOffset());
          assertEquals(1, ackList.get(1).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList.get(1).acknowledgeTypes().get(0));
  
          List<AcknowledgementBatch> ackList2 = acks.getAcknowledgementBatches();
          assertEquals(2, ackList2.size());
          assertEquals(2L, ackList2.get(0).firstOffset());
          assertEquals(2L, ackList2.get(0).lastOffset());
          assertEquals(1, ackList2.get(0).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(0).acknowledgeTypes().get(0));
          assertEquals(4L, ackList2.get(1).firstOffset());
          assertEquals(4L, ackList2.get(1).lastOffset());
          assertEquals(1, ackList2.get(1).acknowledgeTypes().size());
          assertEquals(Acknowledgements.ACKNOWLEDGE_TYPE_GAP, ackList2.get(1).acknowledgeTypes().get(0));
      }
  }
  