You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "jack2012aa (via GitHub)" <gi...@apache.org> on 2025/09/02 16:28:32 UTC
[PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
jack2012aa opened a new pull request, #20461:
URL: https://github.com/apache/kafka/pull/20461
# Description
Admin#desceibeReplicaLogDirs fails all futures when any error happens, regardless of the exception type or the failed directory. Especially, required by the issue, CLUSTER_AUTHORIZATION_FAILED should be handled independently.
# Changes
## Tests
Two tests are added: one for checking whether user can catch the CLUSTER_AUTHORIZATION_FAILED by getCause; one for checking whether failure in unrelated directory will affect other requests.
## CLUSTER_AUTHORIZATION_FAILED
It is handled by an additional if statement now.
## Single Directory-level Error
It is hard to map failed directories back to requested topic partition replica. If a directory or a broker fails, the response won't contain information of any replicas. The new attempt fails replicas that are not mentioned in healthy responses, if there is some error codes. It won't break the behavior of giving an empty result to unreplied replicas if no error code is in response.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] closed pull request #20461: KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly
URL: https://github.com/apache/kafka/pull/20461
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2328841515
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,34 +3109,52 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
Review Comment:
Excuse me, does this PR handle the `DescribeLogDirsResponseData#errorCode`?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "jack2012aa (via GitHub)" <gi...@apache.org>.
jack2012aa commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2329119008
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,34 +3109,52 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
Review Comment:
Thank you. The new version is pushed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "jack2012aa (via GitHub)" <gi...@apache.org>.
jack2012aa commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2323355531
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2517,6 +2517,55 @@ public void testDescribeReplicaLogDirsUnexpected() throws ExecutionException, In
}
}
+ @Test
+ public void testDescribeReplicaLogDirsWithAuthorizationException() throws ExecutionException, InterruptedException {
+ TopicPartitionReplica tpr = new TopicPartitionReplica("topic", 12, 1);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ env.kafkaClient().prepareResponseFrom(
+ prepareDescribeLogDirsResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, broker1log0),
+ env.cluster().nodeById(tpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(singletonList(tpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ Throwable e = assertThrows(Exception.class, () -> values.get(tpr).get());
+ assertInstanceOf(ClusterAuthorizationException.class, e.getCause());
+ }
+
+ }
+
+ @Test
+ public void testDescribeReplicaLogDirsWithSingleDirException() throws ExecutionException, InterruptedException {
+ int brokerId = 1;
+ TopicPartitionReplica successfulTpr = new TopicPartitionReplica("topic", 12, brokerId);
+ TopicPartitionReplica failedTpr = new TopicPartitionReplica("failed", 12, brokerId);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ String broker1log1 = "/var/data/kafka1";
+ int broker1Log0PartitionSize = 987654321;
+ int broker1Log0OffsetLag = 24;
+
+ DescribeLogDirsResponseData.DescribeLogDirsResult successfulResult = prepareDescribeLogDirsResult(
+ successfulTpr, broker1log0, broker1Log0PartitionSize, broker1Log0OffsetLag, false);
+ DescribeLogDirsResponseData.DescribeLogDirsResult failedResult = new DescribeLogDirsResponseData.DescribeLogDirsResult()
+ .setErrorCode(Errors.LOG_DIR_NOT_FOUND.code())
+ .setLogDir(broker1log1);
+ DescribeLogDirsResponse response = new DescribeLogDirsResponse(new DescribeLogDirsResponseData().setResults(asList(successfulResult, failedResult)));
+ env.kafkaClient().prepareResponseFrom(response, env.cluster().nodeById(successfulTpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(asList(successfulTpr, failedTpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ assertNotNull(values.get(successfulTpr).get());
+ assertThrows(Exception.class, () -> values.get(failedTpr).get());
Review Comment:
Yes, an assertion on the exception type and could better ensure that the future ends with the expected reason. They are included in the new commit.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2376370868
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,37 +3108,57 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
- for (Map.Entry<String, LogDirDescription> responseEntry : logDirDescriptions(response).entrySet()) {
+
+ Set<TopicPartition> pendingPartitions = new HashSet<>(replicaDirInfoByPartition.keySet());
+ Map<String, Throwable> directoryFailures = new HashMap<>();
+ Map<String, LogDirDescription> descriptions = logDirDescriptions(response);
+ if (descriptions.isEmpty()) {
+ Errors error = response.data().errorCode() == Errors.NONE.code()
+ ? Errors.CLUSTER_AUTHORIZATION_FAILED
+ : Errors.forCode(response.data().errorCode());
+ handleFailure(error.exception(), pendingPartitions);
+ }
+
+ for (Map.Entry<String, LogDirDescription> responseEntry : descriptions.entrySet()) {
String logDir = responseEntry.getKey();
LogDirDescription logDirInfo = responseEntry.getValue();
// No replica info will be provided if the log directory is offline
if (logDirInfo.error() instanceof KafkaStorageException)
continue;
- if (logDirInfo.error() != null)
- handleFailure(new IllegalStateException(
- "The error " + logDirInfo.error().getClass().getName() + " for log directory " + logDir + " in the response from broker " + brokerId + " is illegal"));
-
- for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
- TopicPartition tp = replicaInfoEntry.getKey();
- ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
- ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
- if (replicaLogDirInfo == null) {
- log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
- } else if (replicaInfo.isFuture()) {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
- replicaLogDirInfo.getCurrentReplicaOffsetLag(),
- logDir,
- replicaInfo.offsetLag()));
- } else {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
- replicaInfo.offsetLag(),
- replicaLogDirInfo.getFutureReplicaLogDir(),
- replicaLogDirInfo.getFutureReplicaOffsetLag()));
+
+ if (logDirInfo.error() == null) {
+ for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
+ TopicPartition tp = replicaInfoEntry.getKey();
+ ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
+ ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
+ if (replicaLogDirInfo == null) {
+ log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
+ } else if (replicaInfo.isFuture()) {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
+ replicaLogDirInfo.getCurrentReplicaOffsetLag(),
+ logDir,
+ replicaInfo.offsetLag()));
+ } else {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
+ replicaInfo.offsetLag(),
+ replicaLogDirInfo.getFutureReplicaLogDir(),
+ replicaLogDirInfo.getFutureReplicaOffsetLag()));
+ }
+ pendingPartitions.remove(tp);
}
+ } else {
+ directoryFailures.put(logDir, logDirInfo.error());
}
}
+ if (!pendingPartitions.isEmpty() && !directoryFailures.isEmpty()) {
Review Comment:
This change is a bit complicated for me. Could you try collecting the topic partitions for each directory error, and then use that collection to complete the `futures`? for example:
```java
final Map<TopicPartition, Exception> failedTps = new HashMap<>();
for (Map.Entry<String, LogDirDescription> responseEntry : logDirDescriptions(response).entrySet()) {
String logDir = responseEntry.getKey();
LogDirDescription logDirInfo = responseEntry.getValue();
// No replica info will be provided if the log directory is offline
if (logDirInfo.error() instanceof KafkaStorageException)
continue;
if (logDirInfo.error() != null)
handleFailure(new IllegalStateException(
"The error " + logDirInfo.error().getClass().getName() + " for log directory " + logDir + " in the response from broker " + brokerId + " is illegal"));
for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
TopicPartition tp = replicaInfoEntry.getKey();
if (logDirInfo.error() != null) {
failedTps.put(tp, logDirInfo.error());
continue;
}
...
}
}
for (var entry : failedTps.entrySet()) {
var tp = entry.getKey();
var future = futures.get(new TopicPartitionReplica(tp.topic(), tp.partition(), brokerId));
future.completeExceptionally(entry.getValue());
}
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,37 +3108,57 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
- for (Map.Entry<String, LogDirDescription> responseEntry : logDirDescriptions(response).entrySet()) {
+
+ Set<TopicPartition> pendingPartitions = new HashSet<>(replicaDirInfoByPartition.keySet());
+ Map<String, Throwable> directoryFailures = new HashMap<>();
+ Map<String, LogDirDescription> descriptions = logDirDescriptions(response);
+ if (descriptions.isEmpty()) {
+ Errors error = response.data().errorCode() == Errors.NONE.code()
+ ? Errors.CLUSTER_AUTHORIZATION_FAILED
+ : Errors.forCode(response.data().errorCode());
+ handleFailure(error.exception(), pendingPartitions);
+ }
+
+ for (Map.Entry<String, LogDirDescription> responseEntry : descriptions.entrySet()) {
String logDir = responseEntry.getKey();
LogDirDescription logDirInfo = responseEntry.getValue();
// No replica info will be provided if the log directory is offline
if (logDirInfo.error() instanceof KafkaStorageException)
continue;
- if (logDirInfo.error() != null)
- handleFailure(new IllegalStateException(
- "The error " + logDirInfo.error().getClass().getName() + " for log directory " + logDir + " in the response from broker " + brokerId + " is illegal"));
-
- for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
- TopicPartition tp = replicaInfoEntry.getKey();
- ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
- ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
- if (replicaLogDirInfo == null) {
- log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
- } else if (replicaInfo.isFuture()) {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
- replicaLogDirInfo.getCurrentReplicaOffsetLag(),
- logDir,
- replicaInfo.offsetLag()));
- } else {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
- replicaInfo.offsetLag(),
- replicaLogDirInfo.getFutureReplicaLogDir(),
- replicaLogDirInfo.getFutureReplicaOffsetLag()));
+
+ if (logDirInfo.error() == null) {
+ for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
+ TopicPartition tp = replicaInfoEntry.getKey();
+ ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
+ ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
+ if (replicaLogDirInfo == null) {
+ log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
+ } else if (replicaInfo.isFuture()) {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
+ replicaLogDirInfo.getCurrentReplicaOffsetLag(),
+ logDir,
+ replicaInfo.offsetLag()));
+ } else {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
+ replicaInfo.offsetLag(),
+ replicaLogDirInfo.getFutureReplicaLogDir(),
+ replicaLogDirInfo.getFutureReplicaOffsetLag()));
+ }
+ pendingPartitions.remove(tp);
}
+ } else {
+ directoryFailures.put(logDir, logDirInfo.error());
}
}
+ if (!pendingPartitions.isEmpty() && !directoryFailures.isEmpty()) {
+ List<String> errorAtDir = new ArrayList<>();
+ directoryFailures.forEach((k, v) -> errorAtDir.add(v.getClass().getName() + " at " + k));
+ Throwable error = new IllegalStateException("The error " + String.join(", ", errorAtDir) + " in the response from broker " + brokerId + " is illegal");
Review Comment:
we could generate the string directly. for example:
```java
directoryFailures.entrySet().stream().map(entry -> entry.getValue().getClass().getName() + " at " + entry.getKey())
.collect(Collectors.joining(","))
```
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20461:
URL: https://github.com/apache/kafka/pull/20461#issuecomment-4921662118
This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "chia7712 (via GitHub)" <gi...@apache.org>.
chia7712 commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2328877321
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,34 +3109,52 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
Review Comment:
> Should I also handle the error in this way?
yes, that looks good to me
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20461:
URL: https://github.com/apache/kafka/pull/20461#issuecomment-4645627175
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "jack2012aa (via GitHub)" <gi...@apache.org>.
jack2012aa commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2328869514
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,34 +3109,52 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
Review Comment:
Sorry, I didn't realize that CLUSTER_AUTHORIZATION_FAILED happens in the response data.
I saw that KafkaAdminClient#describeLogDirs finds the error by testing whether the descriptions map is empty, but not checks the error code directly. Should I also handle the error in this way?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "frankvicky (via GitHub)" <gi...@apache.org>.
frankvicky commented on code in PR #20461:
URL: https://github.com/apache/kafka/pull/20461#discussion_r2322628958
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,34 +3109,54 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
+
+ Set<TopicPartition> pendingPartitions = new HashSet<>(replicaDirInfoByPartition.keySet());
+ Map<String, Throwable> directoryFailures = new HashMap<>();
+
for (Map.Entry<String, LogDirDescription> responseEntry : logDirDescriptions(response).entrySet()) {
String logDir = responseEntry.getKey();
LogDirDescription logDirInfo = responseEntry.getValue();
// No replica info will be provided if the log directory is offline
if (logDirInfo.error() instanceof KafkaStorageException)
continue;
- if (logDirInfo.error() != null)
- handleFailure(new IllegalStateException(
- "The error " + logDirInfo.error().getClass().getName() + " for log directory " + logDir + " in the response from broker " + brokerId + " is illegal"));
-
- for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
- TopicPartition tp = replicaInfoEntry.getKey();
- ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
- ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
- if (replicaLogDirInfo == null) {
- log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
- } else if (replicaInfo.isFuture()) {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
- replicaLogDirInfo.getCurrentReplicaOffsetLag(),
- logDir,
- replicaInfo.offsetLag()));
- } else {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
- replicaInfo.offsetLag(),
- replicaLogDirInfo.getFutureReplicaLogDir(),
- replicaLogDirInfo.getFutureReplicaOffsetLag()));
+ if (logDirInfo.error() instanceof ClusterAuthorizationException)
+ handleFailure(logDirInfo.error());
+
+ if (logDirInfo.error() == null) {
+ for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
+ TopicPartition tp = replicaInfoEntry.getKey();
+ ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
+ ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
+ if (replicaLogDirInfo == null) {
+ log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
+ } else if (replicaInfo.isFuture()) {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
+ replicaLogDirInfo.getCurrentReplicaOffsetLag(),
+ logDir,
+ replicaInfo.offsetLag()));
+ } else {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
+ replicaInfo.offsetLag(),
+ replicaLogDirInfo.getFutureReplicaLogDir(),
+ replicaLogDirInfo.getFutureReplicaOffsetLag()));
+ }
+ pendingPartitions.remove(tp);
}
+ } else {
+ directoryFailures.put(logDir, logDirInfo.error());
+ }
+ }
+
+ if (!pendingPartitions.isEmpty() && !directoryFailures.isEmpty()) {
+ ArrayList<String> errorAtDir = new ArrayList<>();
Review Comment:
nit
```suggestion
List<String> errorAtDir = new ArrayList<>();
```
##########
clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java:
##########
@@ -3108,34 +3109,54 @@ public DescribeLogDirsRequest.Builder createRequest(int timeoutMs) {
@Override
public void handleResponse(AbstractResponse abstractResponse) {
DescribeLogDirsResponse response = (DescribeLogDirsResponse) abstractResponse;
+
+ Set<TopicPartition> pendingPartitions = new HashSet<>(replicaDirInfoByPartition.keySet());
+ Map<String, Throwable> directoryFailures = new HashMap<>();
+
for (Map.Entry<String, LogDirDescription> responseEntry : logDirDescriptions(response).entrySet()) {
String logDir = responseEntry.getKey();
LogDirDescription logDirInfo = responseEntry.getValue();
// No replica info will be provided if the log directory is offline
if (logDirInfo.error() instanceof KafkaStorageException)
continue;
- if (logDirInfo.error() != null)
- handleFailure(new IllegalStateException(
- "The error " + logDirInfo.error().getClass().getName() + " for log directory " + logDir + " in the response from broker " + brokerId + " is illegal"));
-
- for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
- TopicPartition tp = replicaInfoEntry.getKey();
- ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
- ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
- if (replicaLogDirInfo == null) {
- log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
- } else if (replicaInfo.isFuture()) {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
- replicaLogDirInfo.getCurrentReplicaOffsetLag(),
- logDir,
- replicaInfo.offsetLag()));
- } else {
- replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
- replicaInfo.offsetLag(),
- replicaLogDirInfo.getFutureReplicaLogDir(),
- replicaLogDirInfo.getFutureReplicaOffsetLag()));
+ if (logDirInfo.error() instanceof ClusterAuthorizationException)
+ handleFailure(logDirInfo.error());
+
+ if (logDirInfo.error() == null) {
+ for (Map.Entry<TopicPartition, ReplicaInfo> replicaInfoEntry : logDirInfo.replicaInfos().entrySet()) {
+ TopicPartition tp = replicaInfoEntry.getKey();
+ ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
+ ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
+ if (replicaLogDirInfo == null) {
+ log.warn("Server response from broker {} mentioned unknown partition {}", brokerId, tp);
+ } else if (replicaInfo.isFuture()) {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(replicaLogDirInfo.getCurrentReplicaLogDir(),
+ replicaLogDirInfo.getCurrentReplicaOffsetLag(),
+ logDir,
+ replicaInfo.offsetLag()));
+ } else {
+ replicaDirInfoByPartition.put(tp, new ReplicaLogDirInfo(logDir,
+ replicaInfo.offsetLag(),
+ replicaLogDirInfo.getFutureReplicaLogDir(),
+ replicaLogDirInfo.getFutureReplicaOffsetLag()));
+ }
+ pendingPartitions.remove(tp);
}
+ } else {
+ directoryFailures.put(logDir, logDirInfo.error());
+ }
+ }
+
+ if (!pendingPartitions.isEmpty() && !directoryFailures.isEmpty()) {
+ ArrayList<String> errorAtDir = new ArrayList<>();
+ for (Map.Entry<String, Throwable> entry : directoryFailures.entrySet()) {
+ errorAtDir.add(entry.getValue().getClass().getName() + " at " + entry.getKey());
+ }
Review Comment:
nit:
```suggestion
directoryFailures.forEach((k, v) -> errorAtDir.add(v.getClass().getName() + " at " + k));
```
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2517,6 +2517,55 @@ public void testDescribeReplicaLogDirsUnexpected() throws ExecutionException, In
}
}
+ @Test
+ public void testDescribeReplicaLogDirsWithAuthorizationException() throws ExecutionException, InterruptedException {
+ TopicPartitionReplica tpr = new TopicPartitionReplica("topic", 12, 1);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ env.kafkaClient().prepareResponseFrom(
+ prepareDescribeLogDirsResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, broker1log0),
+ env.cluster().nodeById(tpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(singletonList(tpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ Throwable e = assertThrows(Exception.class, () -> values.get(tpr).get());
+ assertInstanceOf(ClusterAuthorizationException.class, e.getCause());
Review Comment:
Please use `TestUtils#assertFutureThrows` instead.
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2517,6 +2517,55 @@ public void testDescribeReplicaLogDirsUnexpected() throws ExecutionException, In
}
}
+ @Test
+ public void testDescribeReplicaLogDirsWithAuthorizationException() throws ExecutionException, InterruptedException {
+ TopicPartitionReplica tpr = new TopicPartitionReplica("topic", 12, 1);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ env.kafkaClient().prepareResponseFrom(
+ prepareDescribeLogDirsResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, broker1log0),
+ env.cluster().nodeById(tpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(singletonList(tpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ Throwable e = assertThrows(Exception.class, () -> values.get(tpr).get());
+ assertInstanceOf(ClusterAuthorizationException.class, e.getCause());
+ }
+
+ }
+
+ @Test
+ public void testDescribeReplicaLogDirsWithSingleDirException() throws ExecutionException, InterruptedException {
+ int brokerId = 1;
+ TopicPartitionReplica successfulTpr = new TopicPartitionReplica("topic", 12, brokerId);
+ TopicPartitionReplica failedTpr = new TopicPartitionReplica("failed", 12, brokerId);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ String broker1log1 = "/var/data/kafka1";
+ int broker1Log0PartitionSize = 987654321;
+ int broker1Log0OffsetLag = 24;
+
+ DescribeLogDirsResponseData.DescribeLogDirsResult successfulResult = prepareDescribeLogDirsResult(
+ successfulTpr, broker1log0, broker1Log0PartitionSize, broker1Log0OffsetLag, false);
+ DescribeLogDirsResponseData.DescribeLogDirsResult failedResult = new DescribeLogDirsResponseData.DescribeLogDirsResult()
+ .setErrorCode(Errors.LOG_DIR_NOT_FOUND.code())
+ .setLogDir(broker1log1);
+ DescribeLogDirsResponse response = new DescribeLogDirsResponse(new DescribeLogDirsResponseData().setResults(asList(successfulResult, failedResult)));
+ env.kafkaClient().prepareResponseFrom(response, env.cluster().nodeById(successfulTpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(asList(successfulTpr, failedTpr));
Review Comment:
List.of
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2517,6 +2517,55 @@ public void testDescribeReplicaLogDirsUnexpected() throws ExecutionException, In
}
}
+ @Test
+ public void testDescribeReplicaLogDirsWithAuthorizationException() throws ExecutionException, InterruptedException {
+ TopicPartitionReplica tpr = new TopicPartitionReplica("topic", 12, 1);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ env.kafkaClient().prepareResponseFrom(
+ prepareDescribeLogDirsResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, broker1log0),
+ env.cluster().nodeById(tpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(singletonList(tpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ Throwable e = assertThrows(Exception.class, () -> values.get(tpr).get());
+ assertInstanceOf(ClusterAuthorizationException.class, e.getCause());
+ }
+
+ }
+
+ @Test
+ public void testDescribeReplicaLogDirsWithSingleDirException() throws ExecutionException, InterruptedException {
+ int brokerId = 1;
+ TopicPartitionReplica successfulTpr = new TopicPartitionReplica("topic", 12, brokerId);
+ TopicPartitionReplica failedTpr = new TopicPartitionReplica("failed", 12, brokerId);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ String broker1log1 = "/var/data/kafka1";
+ int broker1Log0PartitionSize = 987654321;
+ int broker1Log0OffsetLag = 24;
+
+ DescribeLogDirsResponseData.DescribeLogDirsResult successfulResult = prepareDescribeLogDirsResult(
+ successfulTpr, broker1log0, broker1Log0PartitionSize, broker1Log0OffsetLag, false);
+ DescribeLogDirsResponseData.DescribeLogDirsResult failedResult = new DescribeLogDirsResponseData.DescribeLogDirsResult()
+ .setErrorCode(Errors.LOG_DIR_NOT_FOUND.code())
+ .setLogDir(broker1log1);
+ DescribeLogDirsResponse response = new DescribeLogDirsResponse(new DescribeLogDirsResponseData().setResults(asList(successfulResult, failedResult)));
+ env.kafkaClient().prepareResponseFrom(response, env.cluster().nodeById(successfulTpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(asList(successfulTpr, failedTpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ assertNotNull(values.get(successfulTpr).get());
+ assertThrows(Exception.class, () -> values.get(failedTpr).get());
Review Comment:
Do you think we should assert the class of this exception?
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2517,6 +2517,55 @@ public void testDescribeReplicaLogDirsUnexpected() throws ExecutionException, In
}
}
+ @Test
+ public void testDescribeReplicaLogDirsWithAuthorizationException() throws ExecutionException, InterruptedException {
+ TopicPartitionReplica tpr = new TopicPartitionReplica("topic", 12, 1);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ env.kafkaClient().prepareResponseFrom(
+ prepareDescribeLogDirsResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, broker1log0),
+ env.cluster().nodeById(tpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(singletonList(tpr));
Review Comment:
List.of
##########
clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java:
##########
@@ -2517,6 +2517,55 @@ public void testDescribeReplicaLogDirsUnexpected() throws ExecutionException, In
}
}
+ @Test
+ public void testDescribeReplicaLogDirsWithAuthorizationException() throws ExecutionException, InterruptedException {
+ TopicPartitionReplica tpr = new TopicPartitionReplica("topic", 12, 1);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ env.kafkaClient().prepareResponseFrom(
+ prepareDescribeLogDirsResponse(Errors.CLUSTER_AUTHORIZATION_FAILED, broker1log0),
+ env.cluster().nodeById(tpr.brokerId()));
+
+ DescribeReplicaLogDirsResult result = env.adminClient().describeReplicaLogDirs(singletonList(tpr));
+ Map<TopicPartitionReplica, KafkaFuture<DescribeReplicaLogDirsResult.ReplicaLogDirInfo>> values = result.values();
+
+ Throwable e = assertThrows(Exception.class, () -> values.get(tpr).get());
+ assertInstanceOf(ClusterAuthorizationException.class, e.getCause());
+ }
+
+ }
+
+ @Test
+ public void testDescribeReplicaLogDirsWithSingleDirException() throws ExecutionException, InterruptedException {
+ int brokerId = 1;
+ TopicPartitionReplica successfulTpr = new TopicPartitionReplica("topic", 12, brokerId);
+ TopicPartitionReplica failedTpr = new TopicPartitionReplica("failed", 12, brokerId);
+
+ try (AdminClientUnitTestEnv env = mockClientEnv()) {
+ env.kafkaClient().setNodeApiVersions(NodeApiVersions.create());
+ String broker1log0 = "/var/data/kafka0";
+ String broker1log1 = "/var/data/kafka1";
+ int broker1Log0PartitionSize = 987654321;
+ int broker1Log0OffsetLag = 24;
+
+ DescribeLogDirsResponseData.DescribeLogDirsResult successfulResult = prepareDescribeLogDirsResult(
+ successfulTpr, broker1log0, broker1Log0PartitionSize, broker1Log0OffsetLag, false);
+ DescribeLogDirsResponseData.DescribeLogDirsResult failedResult = new DescribeLogDirsResponseData.DescribeLogDirsResult()
+ .setErrorCode(Errors.LOG_DIR_NOT_FOUND.code())
+ .setLogDir(broker1log1);
+ DescribeLogDirsResponse response = new DescribeLogDirsResponse(new DescribeLogDirsResponseData().setResults(asList(successfulResult, failedResult)));
Review Comment:
List.of
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly [kafka]
Posted by "github-actions[bot] (via GitHub)" <gi...@apache.org>.
github-actions[bot] commented on PR #20461:
URL: https://github.com/apache/kafka/pull/20461#issuecomment-3688573823
This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.
<p>
If you are having difficulty finding a reviewer, please reach out on the
[mailing list](https://kafka.apache.org/contact).
<p>
If this PR is no longer valid or desired, please feel free to close it. If no activity
occurs in the next 30 days, it will be automatically closed.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org