This is an automated email from the ASF dual-hosted git repository.
manikumar pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new a9b0fc8 KAFKA-8482: Improve documentation on
AdminClient#alterReplicaLogDirs, AlterReplicaLogDirsResult (#7083)
a9b0fc8 is described below
commit a9b0fc866a02762878a83e21130fdb236766aff5
Author: Lee Dongjin <[email protected]>
AuthorDate: Sun Oct 20 20:43:17 2019 +0900
KAFKA-8482: Improve documentation on AdminClient#alterReplicaLogDirs,
AlterReplicaLogDirsResult (#7083)
Reviewers: Manikumar Reddy <[email protected]>
---
.../java/org/apache/kafka/clients/admin/Admin.java | 16 +++++----
.../clients/admin/AlterReplicaLogDirsResult.java | 42 ++++++++++++++++------
2 files changed, 41 insertions(+), 17 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
index a6458b1..e8bec4a 100644
--- a/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
+++ b/clients/src/main/java/org/apache/kafka/clients/admin/Admin.java
@@ -439,7 +439,7 @@ public interface Admin extends AutoCloseable {
* Change the log directory for the specified replicas. If the replica
does not exist on the broker, the result
* shows REPLICA_NOT_AVAILABLE for the given replica and the replica will
be created in the given log directory on the
* broker when it is created later. If the replica already exists on the
broker, the replica will be moved to the given
- * log directory if it is not already there.
+ * log directory if it is not already there. For detailed result, inspect
the returned {@link AlterReplicaLogDirsResult} instance.
* <p>
* This operation is not transactional so it may succeed for some replicas
while fail for others.
* <p>
@@ -448,8 +448,9 @@ public interface Admin extends AutoCloseable {
* <p>
* This operation is supported by brokers with version 1.1.0 or higher.
*
- * @param replicaAssignment The replicas with their log directory absolute
path
- * @return The AlterReplicaLogDirsResult
+ * @param replicaAssignment The replicas with their log directory
absolute path
+ * @return The AlterReplicaLogDirsResult
+ * @throws InterruptedException Interrupted while joining I/O thread
*/
default AlterReplicaLogDirsResult
alterReplicaLogDirs(Map<TopicPartitionReplica, String> replicaAssignment) {
return alterReplicaLogDirs(replicaAssignment, new
AlterReplicaLogDirsOptions());
@@ -459,15 +460,16 @@ public interface Admin extends AutoCloseable {
* Change the log directory for the specified replicas. If the replica
does not exist on the broker, the result
* shows REPLICA_NOT_AVAILABLE for the given replica and the replica will
be created in the given log directory on the
* broker when it is created later. If the replica already exists on the
broker, the replica will be moved to the given
- * log directory if it is not already there.
+ * log directory if it is not already there. For detailed result, inspect
the returned {@link AlterReplicaLogDirsResult} instance.
* <p>
* This operation is not transactional so it may succeed for some replicas
while fail for others.
* <p>
* This operation is supported by brokers with version 1.1.0 or higher.
*
- * @param replicaAssignment The replicas with their log directory absolute
path
- * @param options The options to use when changing replica dir
- * @return The AlterReplicaLogDirsResult
+ * @param replicaAssignment The replicas with their log directory
absolute path
+ * @param options The options to use when changing replica
dir
+ * @return The AlterReplicaLogDirsResult
+ * @throws InterruptedException Interrupted while joining I/O thread
*/
AlterReplicaLogDirsResult alterReplicaLogDirs(Map<TopicPartitionReplica,
String> replicaAssignment,
AlterReplicaLogDirsOptions
options);
diff --git
a/clients/src/main/java/org/apache/kafka/clients/admin/AlterReplicaLogDirsResult.java
b/clients/src/main/java/org/apache/kafka/clients/admin/AlterReplicaLogDirsResult.java
index 2373265..81eb0ea 100644
---
a/clients/src/main/java/org/apache/kafka/clients/admin/AlterReplicaLogDirsResult.java
+++
b/clients/src/main/java/org/apache/kafka/clients/admin/AlterReplicaLogDirsResult.java
@@ -16,14 +16,24 @@
*/
package org.apache.kafka.clients.admin;
+import java.util.Map;
+import java.util.concurrent.CancellationException;
+import java.util.concurrent.ExecutionException;
import org.apache.kafka.common.KafkaFuture;
import org.apache.kafka.common.TopicPartitionReplica;
import org.apache.kafka.common.annotation.InterfaceStability;
-import java.util.Map;
-
+import org.apache.kafka.common.errors.ClusterAuthorizationException;
+import org.apache.kafka.common.errors.InvalidTopicException;
+import org.apache.kafka.common.errors.KafkaStorageException;
+import org.apache.kafka.common.errors.LogDirNotFoundException;
+import org.apache.kafka.common.errors.ReplicaNotAvailableException;
+import org.apache.kafka.common.errors.UnknownServerException;
/**
* The result of {@link Admin#alterReplicaLogDirs(Map,
AlterReplicaLogDirsOptions)}.
+ *
+ * To retrieve the detailed result per specified {@link
TopicPartitionReplica}, use {@link #values()}. To retrieve the
+ * overall result only, use {@link #all()}.
*/
@InterfaceStability.Evolving
public class AlterReplicaLogDirsResult {
@@ -34,22 +44,34 @@ public class AlterReplicaLogDirsResult {
}
/**
+ * Return a map from {@link TopicPartitionReplica} to {@link KafkaFuture}
which holds the status of individual
+ * replica movement.
*
- * Return a map from replica to future which can be used to check the
status of individual replica movement.
- *
- * Possible error code:
+ * To check the result of individual replica movement, call {@link
KafkaFuture#get()} from the value contained
+ * in the returned map. If there is no error, it will return silently; if
not, an {@link Exception} will be thrown
+ * like the following:
*
- * LOG_DIR_NOT_FOUND (57)
- * KAFKA_STORAGE_ERROR (56)
- * REPLICA_NOT_AVAILABLE (9)
- * UNKNOWN (-1)
+ * <ul>
+ * <li>{@link CancellationException}: The task was canceled.</li>
+ * <li>{@link InterruptedException}: Interrupted while joining I/O
thread.</li>
+ * <li>{@link ExecutionException}: Execution failed with the following
causes:</li>
+ * <ul>
+ * <li>{@link ClusterAuthorizationException}: Authorization failed.
(CLUSTER_AUTHORIZATION_FAILED, 31)</li>
+ * <li>{@link InvalidTopicException}: The specified topic name is too
long. (INVALID_TOPIC_EXCEPTION, 17)</li>
+ * <li>{@link LogDirNotFoundException}: The specified log directory is
not found in the broker. (LOG_DIR_NOT_FOUND, 57)</li>
+ * <li>{@link ReplicaNotAvailableException}: The replica does not
exist on the broker. (REPLICA_NOT_AVAILABLE, 9)</li>
+ * <li>{@link KafkaStorageException}: Disk error occurred.
(KAFKA_STORAGE_ERROR, 56)</li>
+ * <li>{@link UnknownServerException}: Unknown. (UNKNOWN_SERVER_ERROR,
-1)</li>
+ * </ul>
+ * </ul>
*/
public Map<TopicPartitionReplica, KafkaFuture<Void>> values() {
return futures;
}
/**
- * Return a future which succeeds if all the replica movement have
succeeded
+ * Return a {@link KafkaFuture} which succeeds on {@link
KafkaFuture#get()} if all the replica movement have succeeded.
+ * if not, it throws an {@link Exception} described in {@link #values()}
method.
*/
public KafkaFuture<Void> all() {
return KafkaFuture.allOf(futures.values().toArray(new KafkaFuture[0]));
