Shekharrajak commented on code in PR #22835:
URL: https://github.com/apache/kafka/pull/22835#discussion_r3695435648
##########
server-common/src/test/java/org/apache/kafka/server/share/persister/PersisterStateManagerTest.java:
##########
@@ -1534,6 +1539,168 @@ 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
testWriteStateCombinedResponseMissingPartitionFailsOnlyThatHandler() {
+ 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();
+
+ PersisterStateManager.WriteStateHandler present =
batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId, 0);
+ PersisterStateManager.WriteStateHandler missing =
batchedWriteHandler(stateManager, coordinatorNode, groupId, topicId, 1);
+
+ Collection<RequestAndCompletionHandler> requests =
stateManager.generateRequests();
+ assertEquals(1, requests.size());
+
+ // partition 1 is absent from the combined response
+ requests.iterator().next().handler.onComplete(writeStateClientResponse(
+ new WriteShareGroupStateResponseData().setResults(List.of(
+ writeStateResult(topicId, writePartitionResult(0,
Errors.NONE))))));
+
+ assertWriteStateResult(present, topicId, 0, Errors.NONE);
+ assertWriteStateResult(missing, topicId, 1,
Errors.UNKNOWN_SERVER_ERROR);
Review Comment:
the missing partition fails
--
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]
