Repository: apache/kafka
File: core/src/test/java/kafka/server/share/SharePartitionTest.java (Java)

Code before:
        SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();

        // A client acquires 4 batches, 11 -> 20, 21 -> 30, 31 -> 40, 41 -> 50.
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 11), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 21), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 31), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 41), 10);

        // After the acknowledgements, the share partition state will be:
        // 1. 11 -> 20: AVAILABLE
        // 2. 21 -> 30: ACQUIRED
        // 3. 31 -> 40: AVAILABLE
        // 4. 41 -> 50: ACKNOWLEDGED
        sharePartition.acknowledge(MEMBER_ID, List.of(
            new ShareAcknowledgementBatch(11, 20, List.of((byte) 2)),
            new ShareAcknowledgementBatch(31, 40, List.of((byte) 2)),
            new ShareAcknowledgementBatch(41, 50, List.of((byte) 1))
        ));

        // Move the LSO to 41. When the LSO moves ahead, all batches that are AVAILABLE before the new LSO will be ARCHIVED.
        // Thus, the state of the share partition will be:
        // 1. 11 -> 20: ARCHIVED
        // 2. 21 -> 30: ACQUIRED
        // 3. 31 -> 40: ARCHIVED
        // 4. 41 -> 50: ACKNOWLEDGED
        // Note, the records that are in ACQUIRED state will remain in ACQUIRED state and will be transitioned to a Terminal
        // state when the corresponding acquisition lock timer task expires.
        sharePartition.updateCacheAndOffsets(41);

        assertEquals(51, sharePartition.nextFetchOffset());
        assertEquals(41, sharePartition.startOffset());
        assertEquals(50, sharePartition.endOffset());

        assertEquals(4, sharePartition.cachedState().size());
        assertEquals(RecordState.ARCHIVED, sharePartition.cachedState().get(11L).batchState());
        assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(21L).batchState());
        assertEquals(RecordState.ARCHIVED, sharePartition.cachedState().get(31L).batchState());
        assertEquals(RecordState.ACKNOWLEDGED, sharePartition.cachedState().get(41L).batchState());
    }

    @Test
    public void testLsoMovementForArchivingAllAvailableOffsets() {
        SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();

        // A client acquires 4 batches, 11 -> 20, 21 -> 30, 31 -> 40, 41 -> 50.
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 11), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 21), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 31), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 41), 10);

        // After the acknowledgements, the share partition state will be:

Review comment (bug):
In the end of this current method:
Can you call `sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(21L, 30L, List.of((byte) 2))))`. Validate the response and cached state. This will make sure that acknowledge did nothing and also didn't fail. However, I ll discuss this behaviour in a separate PR, when I ll update the method docs.

Then call `sharePartition.cachedState().get(21L).batchAcquisitionLockTimeoutTask().run();` and validate the respective records are directly archived.

Code after:
        SharePartition sharePartition = SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();

        // A client acquires 4 batches, 11 -> 20, 21 -> 30, 31 -> 40, 41 -> 50.
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 11), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 21), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 31), 10);
        fetchAcquiredRecords(sharePartition, memoryRecords(10, 41), 10);

        // After the acknowledgements, the state of share partition will be:
        // 1. 11 -> 20: AVAILABLE
        // 2. 21 -> 30: ACQUIRED
        // 3. 31 -> 40: AVAILABLE
        // 4. 41 -> 50: ACQUIRED
        sharePartition.acknowledge(MEMBER_ID, List.of(
            new ShareAcknowledgementBatch(11, 20, List.of((byte) 2)),
            new ShareAcknowledgementBatch(31, 40, List.of((byte) 2))
        ));

        // Move the LSO to 41. When the LSO moves ahead, all batches that are AVAILABLE before the new LSO will be ARCHIVED.
        // Thus, the state of the share partition will be:
        // 1. 11 -> 20: ARCHIVED
        // 2. 21 -> 30: ACQUIRED
        // 3. 31 -> 40: ARCHIVED
        // 4. 41 -> 50: ACQUIRED
        // Note, the records that are in ACQUIRED state will remain in ACQUIRED state and will be transitioned to a Terminal
        // state when the corresponding acquisition lock timer task expires.
        sharePartition.updateCacheAndOffsets(41);

        assertEquals(51, sharePartition.nextFetchOffset());
        assertEquals(41, sharePartition.startOffset());
        assertEquals(50, sharePartition.endOffset());

        assertEquals(4, sharePartition.cachedState().size());
        assertEquals(RecordState.ARCHIVED, sharePartition.cachedState().get(11L).batchState());
        assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(21L).batchState());
        assertEquals(RecordState.ARCHIVED, sharePartition.cachedState().get(31L).batchState());
        assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(41L).batchState());

        // The client acknowledges the batch 21 -> 30. Since this batch is before the LSO, nothing will be done and these
        // records will remain in the ACQUIRED state.
        sharePartition.acknowledge(MEMBER_ID, List.of(new ShareAcknowledgementBatch(21L, 30L, List.of((byte) 2))));

        // The batch is still in ACQUIRED state.
        assertEquals(RecordState.ACQUIRED, sharePartition.cachedState().get(21L).batchState());

        // Once the acquisition lock timer task for the batch 21 -> 30 is expired, these records will directly be
        // ARCHIVED.
        sharePartition.cachedState().get(21L).batchAcquisitionLockTimeoutTask().run();
        assertEquals(RecordState.ARCHIVED, sharePartition.cachedState().get(21L).batchState());
    }