Shekharrajak commented on code in PR #22835:
URL: https://github.com/apache/kafka/pull/22835#discussion_r3695492288
##########
server-common/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java:
##########
@@ -1534,6 +1540,223 @@ public void
testWriteStateRequestBatchingWithCoordinatorNodeLookup() throws Exec
TestUtils.waitForCondition(isBatchingSuccess::get,
TestUtils.DEFAULT_MAX_WAIT_MS, 10L, () -> "unable to verify batching");
}
+ private static WriteShareGroupStateResponseData.PartitionResult
writePartitionResult(int partition, Errors error) {
+ return new WriteShareGroupStateResponseData.PartitionResult()
+ .setPartition(partition)
+ .setErrorCode(error.code())
+ .setErrorMessage(error.message());
+ }
+
+ private static WriteShareGroupStateResponseData.WriteStateResult
writeStateResult(
+ Uuid topicId,
+ WriteShareGroupStateResponseData.PartitionResult... partitions
+ ) {
+ return new WriteShareGroupStateResponseData.WriteStateResult()
+ .setTopicId(topicId)
+ .setPartitions(List.of(partitions));
+ }
+
+ private static ClientResponse
writeStateClientResponse(WriteShareGroupStateResponseData data) {
+ return new ClientResponse(
+ new RequestHeader(ApiKeys.WRITE_SHARE_GROUP_STATE,
ApiKeys.WRITE_SHARE_GROUP_STATE.latestVersion(), "client-id", 0),
+ null,
+ HOST,
+ MOCK_TIME.milliseconds(),
+ MOCK_TIME.milliseconds(),
+ false,
+ null,
+ null,
+ new WriteShareGroupStateResponse(data)
+ );
+ }
+
+ // Places a write handler straight into the batching map, bypassing
coordinator lookup, so that a
+ // single generateRequests() call yields one coalesced request covering
every handler.
+ private PersisterStateManager.WriteStateHandler batchedWriteHandler(
+ PersisterStateManager stateManager,
+ Node coordinatorNode,
+ String groupId,
+ Uuid topicId,
+ int partition
+ ) {
+ PersisterStateManager.WriteStateHandler handler = stateManager.new
WriteStateHandler(
+ groupId,
+ topicId,
+ partition,
+ 0,
+ 0,
+ 0,
+ 0,
+ List.of(),
+ new CompletableFuture<>(),
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_RPC_RETRY_ATTEMPTS
+ );
+ handler.addRequestToNodeMap(coordinatorNode, handler);
+ return handler;
+ }
+
+ private static void assertWriteStateResult(
+ PersisterStateManager.WriteStateHandler handler,
+ Uuid expectedTopicId,
+ int expectedPartition,
+ Errors expectedError
+ ) {
+ WriteShareGroupStateResponse response = assertDoesNotThrow(() ->
handler.result().get());
+ assertEquals(1, response.data().results().size());
+ WriteShareGroupStateResponseData.WriteStateResult result =
response.data().results().get(0);
+ assertEquals(expectedTopicId, result.topicId());
+ assertEquals(1, result.partitions().size());
+ assertEquals(expectedPartition,
result.partitions().get(0).partition());
+ assertEquals(expectedError.code(),
result.partitions().get(0).errorCode());
+ }
+
+ @Test
+ public void testWriteStateCombinedResponseDemultiplexedPerTopicPartition()
{
+ String groupId = "group1";
+ Uuid topicId1 = Uuid.randomUuid();
+ Uuid topicId2 = Uuid.randomUuid();
+ Node coordinatorNode = new Node(1, HOST, PORT);
+
+ PersisterStateManager stateManager =
PersisterStateManagerBuilder.builder()
+ .withKafkaClient(new MockClient(MOCK_TIME))
+ .withTimer(mockTimer)
+ .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode))
+ .build();
+
+ // the same partition numbers appear under both topics, so a
topicId-blind lookup cannot pass
+ PersisterStateManager.WriteStateHandler topic1Partition0 =
batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId1, 0);
+ PersisterStateManager.WriteStateHandler topic1Partition1 =
batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId1, 1);
+ PersisterStateManager.WriteStateHandler topic2Partition0 =
batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId2, 0);
+ PersisterStateManager.WriteStateHandler topic2Partition1 =
batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId2, 1);
+
+ Collection<RequestAndCompletionHandler> requests =
stateManager.generateRequests();
+ assertEquals(1, requests.size());
+
+ // a distinct error per (topic, partition) makes any mixed-up slice
observable
+ requests.iterator().next().handler.onComplete(writeStateClientResponse(
+ new WriteShareGroupStateResponseData().setResults(List.of(
+ writeStateResult(topicId1,
+ writePartitionResult(0, Errors.NONE),
+ writePartitionResult(1, Errors.INVALID_REQUEST)),
+ writeStateResult(topicId2,
+ writePartitionResult(0, Errors.FENCED_STATE_EPOCH),
+ writePartitionResult(1, Errors.NONE))))));
+
+ assertWriteStateResult(topic1Partition0, topicId1, 0, Errors.NONE);
+ assertWriteStateResult(topic1Partition1, topicId1, 1,
Errors.INVALID_REQUEST);
+ assertWriteStateResult(topic2Partition0, topicId2, 0,
Errors.FENCED_STATE_EPOCH);
+ assertWriteStateResult(topic2Partition1, topicId2, 1, Errors.NONE);
+ }
+
+ @Test
+ public void testWriteStateCombinedResponseIndexedOncePerBatch() {
+ String groupId = "group1";
+ Uuid topicId = Uuid.randomUuid();
+ Node coordinatorNode = new Node(1, HOST, PORT);
+
+ PersisterStateManager stateManager =
PersisterStateManagerBuilder.builder()
+ .withKafkaClient(new MockClient(MOCK_TIME))
+ .withTimer(mockTimer)
+ .withCacheHelper(getCoordinatorCacheHelper(coordinatorNode))
+ .build();
+
+ List<PersisterStateManager.WriteStateHandler> handlers = new
ArrayList<>();
+ for (int partition = 0; partition < 4; partition++) {
+ PersisterStateManager.WriteStateHandler handler =
spy(stateManager.new WriteStateHandler(
+ groupId,
+ topicId,
+ partition,
+ 0,
+ 0,
+ 0,
+ 0,
+ List.of(),
+ new CompletableFuture<>(),
+ REQUEST_BACKOFF_MS,
+ REQUEST_BACKOFF_MAX_MS,
+ MAX_RPC_RETRY_ATTEMPTS
+ ));
+ handler.addRequestToNodeMap(coordinatorNode, handler);
+ handlers.add(handler);
+ }
+
+ Collection<RequestAndCompletionHandler> requests =
stateManager.generateRequests();
+ assertEquals(1, requests.size());
+
+ requests.iterator().next().handler.onComplete(writeStateClientResponse(
+ new WriteShareGroupStateResponseData().setResults(List.of(
+ writeStateResult(topicId,
+ writePartitionResult(0, Errors.NONE),
+ writePartitionResult(1, Errors.NONE),
+ writePartitionResult(2, Errors.NONE),
+ writePartitionResult(3, Errors.NONE))))));
+
+ // the combined response is indexed once for the whole batch, and
every other handler reuses
+ // that index rather than rescanning the response
+ verify(handlers.get(0), times(1)).buildResultIndex(any());
+ for (int i = 1; i < handlers.size(); i++) {
+ verify(handlers.get(i), never()).buildResultIndex(any());
Review Comment:
If sharedResultIndex was not handed to it, it is forced into that fallback
and builds its own index. Here
https://github.com/apache/kafka/pull/22835/changes#diff-14030e8d09d19bdb8e5b37a07657bef62db5cb9d76dce9bb0ebcf3058f6fd002R319
So one call on handler 0, zero across the other handlers
--
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]
