Skip to content

KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly #20461

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 10 commits into
base: trunk
Choose a base branch
from
+261 −192
Open

KAFKA-19631 Admin#describeReplicaLogDirs does not handle error correctly #20461

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
10 commits
Select commit Hold shift + click to select a range
0afc129
Add a test for cluster authorization failure
jack2012aa Aug 25, 2025
96dfa85
Add a test for single directory-level error
jack2012aa Sep 1, 2025
46949fc
Propagate CLUSTER_AUTHORIZATION_FAILED
jack2012aa Sep 2, 2025
cd93d18
Handle failure in directory-granularity
jack2012aa Sep 2, 2025
6cd8f3b
Merge branch 'apache:trunk' into KAFKA-19631
jack2012aa Sep 2, 2025
4fe6399
Fix format violation
jack2012aa Sep 3, 2025
3213695
Refactor assertThrows to assertFutureThrows
jack2012aa Sep 4, 2025
6b837c0
Refactor asList and singletonList to List.of
jack2012aa Sep 4, 2025
92fde8b
Refactor list and map
jack2012aa Sep 4, 2025
96031a4
Add a test on exception message
jack2012aa Sep 4, 2025
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
59 changes: 39 additions & 20 deletions clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -91,6 +91,7 @@
import org.apache.kafka.common.config.ConfigResource;
import org.apache.kafka.common.errors.ApiException;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.ClusterAuthorizationException;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.errors.InvalidRequestException;
import org.apache.kafka.common.errors.InvalidTopicException;
Expand Down Expand Up @@ -3108,34 +3109,52 @@ 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()) {
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");
for (TopicPartition tp: pendingPartitions) {
KafkaFutureImpl<ReplicaLogDirInfo> future = futures.get(new TopicPartitionReplica(tp.topic(), tp.partition(), brokerId));
future.completeExceptionally(error);
}
}

Expand Down
Loading