Shekharrajak commented on code in PR #22357:
URL: https://github.com/apache/kafka/pull/22357#discussion_r3409733564
##########
core/src/test/java/kafka/server/share/SharePartitionTest.java:
##########
@@ -13305,6 +13350,286 @@ private static ShareGroupConfigProvider
configProviderWithRenewDisabled() {
return configProvider;
}
+ @Test
+ public void
testStageTxnAcknowledgeAcceptThenCommitMarkerTransitionsToAcknowledged() {
+ SharePartition sharePartition =
SharePartitionBuilder.builder().withState(SharePartitionState.ACTIVE).build();
+ fetchAcquiredRecords(sharePartition, memoryRecords(10, 5), 5);
+
+ CompletableFuture<Void> stage =
sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.ACCEPT.id))));
+ assertNull(stage.join());
+ assertEquals(RecordState.TX_PENDING,
sharePartition.cachedState().get(10L).batchState());
+
+ sharePartition.applyTxnMarker(100L, (short) 1,
TransactionResult.COMMIT);
+ assertTrue(sharePartition.cachedState().isEmpty());
+ assertEquals(15, sharePartition.startOffset());
+ }
+
+ @Test
+ public void testStageTxnAcknowledgePersistsPendingAckMetadata() {
+ Persister persister = Mockito.mock(Persister.class);
+ Mockito.when(persister.writeState(Mockito.any()))
+
.thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult(Errors.NONE)));
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withPersister(persister)
+ .withState(SharePartitionState.ACTIVE)
+ .build();
+ fetchAcquiredRecords(sharePartition, memoryRecords(10, 5), 5);
+
+ sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.ACCEPT.id)))).join();
+
+ ArgumentCaptor<WriteShareGroupStateParameters> captor =
+ ArgumentCaptor.forClass(WriteShareGroupStateParameters.class);
+ Mockito.verify(persister,
Mockito.times(1)).writeState(captor.capture());
+ PartitionStateBatchData partitionData = captor.getValue()
+ .groupTopicPartitionData()
+ .topicsData()
+ .get(0)
+ .partitions()
+ .get(0);
+ PersisterStateBatch stateBatch = partitionData.stateBatches().get(0);
+ assertEquals(10L, stateBatch.firstOffset());
+ assertEquals(14L, stateBatch.lastOffset());
+ assertEquals(RecordState.TX_PENDING.id(), stateBatch.deliveryState());
+ assertEquals((short) 1, stateBatch.deliveryCount());
+ assertEquals(100L, stateBatch.stagedProducerId());
+ assertEquals((short) 1, stateBatch.stagedProducerEpoch());
+ assertEquals(AcknowledgeType.ACCEPT.id, stateBatch.stagedAckType());
+ }
+
+ @Test
+ public void testStageTxnAcknowledgeRetryWithSameAckIsIdempotent() {
+ Persister persister = Mockito.mock(Persister.class);
+ Mockito.when(persister.writeState(Mockito.any()))
+
.thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult(Errors.NONE)));
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withPersister(persister)
+ .withState(SharePartitionState.ACTIVE)
+ .build();
+ fetchAcquiredRecords(sharePartition, memoryRecords(10, 5), 5);
+
+ sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.ACCEPT.id)))).join();
+ sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.ACCEPT.id)))).join();
+
+ assertEquals(RecordState.TX_PENDING,
sharePartition.cachedState().get(10L).batchState());
+ assertEquals(AcknowledgeType.ACCEPT.id,
sharePartition.cachedState().get(10L).batchStagedAckType());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
+ }
+
+ @Test
+ public void
testStageTxnAcknowledgeRetryWithDifferentAckFailsWithoutRollback() {
+ Persister persister = Mockito.mock(Persister.class);
+ Mockito.when(persister.writeState(Mockito.any()))
+
.thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult(Errors.NONE)));
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withPersister(persister)
+ .withState(SharePartitionState.ACTIVE)
+ .build();
+ fetchAcquiredRecords(sharePartition, memoryRecords(10, 5), 5);
+
+ sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.ACCEPT.id)))).join();
+ CompletableFuture<Void> conflictingRetry =
sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.REJECT.id))));
+
+ assertFutureThrows(InvalidRecordStateException.class,
conflictingRetry);
+ assertEquals(RecordState.TX_PENDING,
sharePartition.cachedState().get(10L).batchState());
+ assertEquals(AcknowledgeType.ACCEPT.id,
sharePartition.cachedState().get(10L).batchStagedAckType());
+ Mockito.verify(persister, Mockito.times(1)).writeState(Mockito.any());
+ }
+
+ @Test
+ public void testStageTxnAcknowledgeRevertsPendingAckOnPersistFailure() {
+ Persister persister = Mockito.mock(Persister.class);
+ Mockito.when(persister.writeState(Mockito.any()))
+
.thenReturn(CompletableFuture.completedFuture(writeShareGroupStateResult(Errors.GROUP_ID_NOT_FOUND)));
+ SharePartition sharePartition = SharePartitionBuilder.builder()
+ .withPersister(persister)
+ .withState(SharePartitionState.ACTIVE)
+ .build();
+ fetchAcquiredRecords(sharePartition, memoryRecords(10, 5), 5);
+
+ CompletableFuture<Void> stage =
sharePartition.stageTxnAcknowledge(MEMBER_ID, 100L, (short) 1,
+ List.of(new ShareAcknowledgementBatch(10, 14,
List.of(AcknowledgeType.ACCEPT.id))));
+
+ assertFutureThrows(GroupIdNotFoundException.class, stage);
+ assertEquals(RecordState.ACQUIRED,
sharePartition.cachedState().get(10L).batchState());
+ assertEquals(-1L,
sharePartition.cachedState().get(10L).batchStagedProducerId());
+ }
+
+ @Test
+ public void testApplyTxnMarkerPersistsFinalState() {
Review Comment:
pending accepted records become ACKNOWLEDGED.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
