Shekharrajak commented on code in PR #22835:
URL: https://github.com/apache/kafka/pull/22835#discussion_r3695461543
##########
server-common/src/main/java/org/apache/kafka/server/share/persister/PersisterStateManager.java:
##########
@@ -783,55 +834,44 @@ protected void handleRequestResponse(ClientResponse
response) {
String clientResponseErrorMessage = clientResponseError.message();
switch (clientResponseError) {
case NONE:
- // response can be a combined one for large number of
requests
- // we need to deconstruct it
- WriteShareGroupStateResponse combinedResponse =
(WriteShareGroupStateResponse) response.responseBody();
-
- for (WriteShareGroupStateResponseData.WriteStateResult
writeStateResult : combinedResponse.data().results()) {
- if
(writeStateResult.topicId().equals(partitionKey().topicId())) {
-
Optional<WriteShareGroupStateResponseData.PartitionResult> partitionStateData =
-
writeStateResult.partitions().stream().filter(partitionResult ->
partitionResult.partition() == partitionKey().partition())
- .findFirst();
-
- if (partitionStateData.isPresent()) {
- Errors error =
Errors.forCode(partitionStateData.get().errorCode());
- String errorMessage =
partitionStateData.get().errorMessage();
- if (errorMessage == null ||
errorMessage.isEmpty()) {
- errorMessage = error.message();
- }
-
- switch (error) {
- case NONE:
- writeStateBackoff.resetAttempts();
-
WriteShareGroupStateResponseData.WriteStateResult result =
WriteShareGroupStateResponse.toResponseWriteStateResult(
- partitionKey().topicId(),
- List.of(partitionStateData.get())
- );
- this.result.complete(new
WriteShareGroupStateResponse(
- new
WriteShareGroupStateResponseData().setResults(List.of(result))));
- return;
-
- // check retriable errors
- case COORDINATOR_NOT_AVAILABLE:
- case COORDINATOR_LOAD_IN_PROGRESS:
- case NOT_COORDINATOR:
- case UNKNOWN_TOPIC_OR_PARTITION:
- log().debug("Received retriable error
in write state RPC for key {}: {}", partitionKey(), errorMessage);
- if (!writeStateBackoff.canAttempt()) {
- log().error("Exhausted max retries
for write state RPC for key {} without success.", partitionKey());
- requestErrorResponse(error, new
Exception("Exhausted max retries to complete write state RPC without
success."));
- return;
- }
- super.resetCoordinatorNode();
- timer.add(new
PersisterTimerTask(writeStateBackoff.backOff(), this));
- return;
+ WriteShareGroupStateResponseData.PartitionResult
partitionResult = lookupPartitionResult(response);
Review Comment:
About heap : lookupPartitionResult drops each handler's reference once it is
consumed. so it will be alive for one callback only. No memory leak.
--
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]
