KAFKA-10120: Deprecate DescribeLogDirsResult.all() and .values() #9007

tombentley · 2020-07-10T13:54:39Z

As per KIP-621. Also added some tests in KafkaAdminClientTest

tombentley · 2020-07-10T13:57:34Z

@cmccabe, @omkreddy, @mimaison, @dajac, @dongjinleekr since you all voted on the KIP feel free to review.

dajac

@tombentley Thanks for the PR. It looks good overall. I have left few comments.

dajac · 2020-07-10T14:50:29Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+        return "LogDirDescription{" +
+                "replicaInfos=" + replicaInfos +
+                ", error=" + error +
+                '}';


nit: We usually use parenthesis instead of curly braces.

dajac · 2020-07-10T14:50:49Z

clients/src/main/java/org/apache/kafka/clients/admin/ReplicaInfo.java

+        return "ReplicaInfo{" +
+                "size=" + size +
+                ", offsetLag=" + offsetLag +
+                ", isFuture=" + isFuture +
+                '}';


nit: curly braces instead of parenthesis.

dajac · 2020-07-10T14:52:32Z

core/src/main/scala/kafka/admin/LogDirsCommand.scala

@@ -64,7 +63,7 @@ object LogDirsCommand {
                    "logDirs" -> logDirInfos.map { case (logDir, logDirInfo) =>
                        Map(
                            "logDir" -> logDir,
-                            "error" -> logDirInfo.error.exceptionName(),
+                            "error" -> Option(logDirInfo.error).flatMap(ex => Some(ex.getClass.getName)).orNull,


nit: Can't we use map instead of flatMap and remove the Some?

dajac · 2020-07-10T15:20:46Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

+        HashMap<String, LogDirDescription> result = new HashMap<>(response.data().results().size());
+        for (DescribeLogDirsResponseData.DescribeLogDirsResult logDirResult : response.data().results()) {
+            Map<TopicPartition, ReplicaInfo> replicaInfoMap = new HashMap<>();
+            if (logDirResult.topics() != null) {


nit: topics is not nullable in the protocol so it should never be null, does it?

dajac · 2020-07-10T15:22:49Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

                            TopicPartition tp = replicaInfoEntry.getKey();
-                            DescribeLogDirsResponse.ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
+                            ReplicaInfo replicaInfo = replicaInfoEntry.getValue();
                            ReplicaLogDirInfo replicaLogDirInfo = replicaDirInfoByPartition.get(tp);
                            if (replicaLogDirInfo == null) {
                                handleFailure(new IllegalStateException(


Not related to your PR but this look weird. It seems that we fail all the futures if an unexpected replica is provided by the broker in the response. I think that we should log a warning when this happen like we do in the other methods (e.g. createTopics). What do you think?

dajac · 2020-07-10T15:25:46Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+            Map<Integer, KafkaFuture<Map<String, LogDirDescription>>> descriptions = result.descriptions();
+            assertEquals(Collections.singleton(0), descriptions.keySet());
+            assertNotNull(descriptions.get(0));
+            assertEquals(Collections.singleton("/var/data/kafka"), descriptions.get(0).get().keySet());
+            assertNull(descriptions.get(0).get().get("/var/data/kafka").error());
+            assertEquals(Collections.singleton(tp), descriptions.get(0).get().get("/var/data/kafka").replicaInfos().keySet());
+            assertEquals(1234567890, descriptions.get(0).get().get("/var/data/kafka").replicaInfos().get(tp).size());
+            assertEquals(0, descriptions.get(0).get().get("/var/data/kafka").replicaInfos().get(tp).offsetLag());
+            assertFalse(descriptions.get(0).get().get("/var/data/kafka").replicaInfos().get(tp).isFuture());


These blocks of assertions are quite hard to read. Can we try to make them more digestable? We could perhaps extract temporary variable to reduce the number of .get(). We could also define an verifyDescription helper that verify a LogDirDescription for instance. It may be worth having dedicated unit tests for the new and the old APIs as well.

dajac · 2020-07-10T17:31:24Z

core/src/test/scala/unit/kafka/server/DescribeLogDirsRequestTest.scala

+    case class ReplicaInfo(size: Long, offsetLag: Long, isFuture: Boolean)
+    case class LogDirInfo(error: Errors, replicaInfos: Map[TopicPartition, ReplicaInfo])


That's a pity that we have to redefine there classes here. Couldn't we update the test to work with the plain response instead?

tombentley · 2020-07-13T09:26:33Z

Thats for the review @dajac. I've addressed your comments.

dongjinleekr · 2020-07-14T13:48:58Z

core/src/main/scala/kafka/admin/LogDirsCommand.scala

@@ -64,7 +63,7 @@ object LogDirsCommand {
                    "logDirs" -> logDirInfos.map { case (logDir, logDirInfo) =>
                        Map(
                            "logDir" -> logDir,
-                            "error" -> logDirInfo.error.exceptionName(),
+                            "error" -> Option(logDirInfo.error).map(ex => ex.getClass.getName).orNull,


+1 for the Option way. 😄

dongjinleekr

LGTM. One minor change! 😃

dongjinleekr · 2020-07-14T14:23:04Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+     * possibly some other exception if there were problems describing the log directory
+     * or null if the directory is online.
+     */
+    public ApiException error() {


How about this? (in consistency with current DescribeLogDirsResponse.LogDirInfo):

Returns `ApiException` if the log directory is offline or an error occurred. If not, returns null. <ul> <li> KafkaStorageException - The log directory is offline. <li> UnknownServerException - The server experienced an unexpected error when processing the request. </ul>

(Description of UnknownServerException was from Errors.UNKNOWN_SERVER_ERROR)

For consistency with the other Javadoc it seems like we should use a single  only when between the paragraphs. Please remove the  tags. (Sorry, I was also confused.)

We should remove the  before the <ul> also. (see below.)

…alues() As per KIP-621. Also added some tests in KafkaAdminClientTest

tombentley · 2020-07-14T15:21:53Z

@dongjinleekr thanks for the review, amended. Also rebased for conflict.

chia7712 · 2020-07-14T16:22:41Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+    private final Map<TopicPartition, ReplicaInfo> replicaInfos;
+    private final ApiException error;
+
+    public LogDirDescription(ApiException error, Map<TopicPartition, ReplicaInfo> replicaInfos) {


Is it better to have package-private visibility?

I initially thought the same, but TopicDescription, for example (as well as other classes accessible from *Results classes) have a public constructors.

That's great point. At the moment, I think that we are not consistent about this. Some are package private and some are not. The advantage of keeping it public is that it allows to use the class in unit tests which resides in other packages.

chia7712 · 2020-07-14T16:32:14Z

clients/src/main/java/org/apache/kafka/common/requests/DescribeLogDirsResponse.java

@@ -95,6 +76,7 @@ public static DescribeLogDirsResponse parse(ByteBuffer buffer, short version) {
     * KAFKA_STORAGE_ERROR (56)
     * UNKNOWN (-1)
     */
+    @Deprecated


Does it need comment to describe the replacement? for example

@deprecated Deprecated Since Kafka 2.7. Use {@link LogDirDescription}.

dajac

@tombentley Thanks for the updated PR. I have left some more comments.

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

dajac · 2020-07-14T18:46:17Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+    private final Map<TopicPartition, ReplicaInfo> replicaInfos;
+    private final ApiException error;
+
+    public LogDirDescription(ApiException error, Map<TopicPartition, ReplicaInfo> replicaInfos) {


That's great point. At the moment, I think that we are not consistent about this. Some are package private and some are not. The advantage of keeping it public is that it allows to use the class in unit tests which resides in other packages.

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

dajac · 2020-07-14T19:13:23Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+            assertEquals(Collections.singleton("/var/data/kafka"), descriptionsMap.keySet());
+            assertNull(descriptionsMap.get("/var/data/kafka").error());
+            Map<TopicPartition, ReplicaInfo> descriptionsReplicaInfos = descriptionsMap.get("/var/data/kafka").replicaInfos();
+            assertEquals(Collections.singleton(tp), descriptionsReplicaInfos.keySet());
+            assertEquals(1234567890, descriptionsReplicaInfos.get(tp).size());
+            assertEquals(0, descriptionsReplicaInfos.get(tp).offsetLag());
+            assertFalse(descriptionsReplicaInfos.get(tp).isFuture());


This block of assertions is used multiple times. Would it make sense to extract it in a helper method, say assertDescriptions, that verifies a descriptions map contains the information about a single log dir/topic partition?

Something like assertDescriptionContains(descriptionsMap, logDir, tp, size, offsetLag, isFuture).

This might have less mileage than you expected because the different types mean we need two methods each with two call sites, rather than 4 call sites for a single method, but I've done it anyway.

dajac · 2020-07-14T19:15:56Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+    }
+
+    @Test
+    public void testDescribeReplicaLogDirs() throws ExecutionException, InterruptedException {


This is not due to your PR but shall we add a unit test which uses multiple brokers?

I added it to the existing test. Due to the new helper methods I felt this didn't really complicate the test very much and is also allows us to cover the case where the RPC returns STORAGE_ERROR.

dongjinleekr · 2020-07-15T12:39:27Z

clients/src/main/java/org/apache/kafka/clients/admin/LogDirDescription.java

+     * possibly some other exception if there were problems describing the log directory
+     * or null if the directory is online.
+     */
+    public ApiException error() {


For consistency with the other Javadoc it seems like we should use a single  only when between the paragraphs. Please remove the  tags. (Sorry, I was also confused.)

tombentley · 2020-07-15T12:47:35Z

@dongjinleekr ah, thank you, I hadn't noticed that that was the norm.

dajac

@tombentley Thanks for the update. LGTM pending jenkins.

omkreddy · 2020-07-23T04:59:26Z

ok to test

mimaison

Thanks @tombentley. I left a few comments

mimaison · 2020-07-23T15:34:05Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -2306,6 +2310,22 @@ void handleFailure(Throwable throwable) {
        return new DescribeLogDirsResult(new HashMap<>(futures));
    }

+    private Map<String, LogDirDescription> logDirDescriptions(DescribeLogDirsResponse response) {


This can be static. Also should we keep it in DescribeLogDirsResponse?

mimaison · 2020-07-23T15:34:24Z

clients/src/main/java/org/apache/kafka/clients/admin/KafkaAdminClient.java

@@ -2306,6 +2310,22 @@ void handleFailure(Throwable throwable) {
        return new DescribeLogDirsResult(new HashMap<>(futures));
    }

+    private Map<String, LogDirDescription> logDirDescriptions(DescribeLogDirsResponse response) {
+        HashMap<String, LogDirDescription> result = new HashMap<>(response.data().results().size());


The left side can be Map

mimaison · 2020-07-23T15:59:23Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

@@ -1057,6 +1059,263 @@ public void testDescribeConfigsUnrequested() throws Exception {
        }
    }

+    private DescribeLogDirsResponse prepareDescribeLogDirsResponse(Errors error, String logDir, TopicPartition tp, long partitionSize, long offsetLag) {


This can be static

mimaison · 2020-07-23T15:59:33Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+                prepareDescribeLogDirsTopics(partitionSize, offsetLag, tp.topic(), tp.partition(), false));
+    }
+
+    private List<DescribeLogDirsTopic> prepareDescribeLogDirsTopics(


This can be static

mimaison · 2020-07-23T16:00:17Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+                        .setOffsetLag(offsetLag))));
+    }
+
+    private DescribeLogDirsResponse prepareDescribeLogDirsResponse(Errors error, String logDir,


This can be static

mimaison · 2020-07-23T16:04:57Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+        }
+    }
+
+    private void assertDescriptionContains(Map<String, LogDirDescription> descriptionsMap, String logDir,


This can be static

mimaison · 2020-07-23T16:05:09Z

clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java

+    }
+
+    @SuppressWarnings("deprecation")
+    private void assertDescriptionContains(Map<String, DescribeLogDirsResponse.LogDirInfo> descriptionsMap,


This can be static

tombentley · 2020-07-28T07:46:16Z

@mimaison done, thanks.

mimaison · 2020-07-30T08:54:46Z

retest this please

mimaison · 2020-07-30T09:27:43Z

ok to test

mimaison · 2020-07-30T13:07:26Z

Failures look unrelated:

org.apache.kafka.streams.integration.EosBetaUpgradeIntegrationTest > shouldUpgradeFromEosAlphaToEosBeta[true] FAILED
org.apache.kafka.connect.mirror.MirrorConnectorsIntegrationTest > testOneWayReplicationWithAutorOffsetSync1

dongjinleekr · 2020-08-01T06:46:43Z

@tombentley Congratulations! ㊗️ @omkreddy @mimaison Thanks again for the detailed review, as usual! 😃

dajac reviewed Jul 10, 2020

View reviewed changes

dongjinleekr reviewed Jul 14, 2020

View reviewed changes

dongjinleekr suggested changes Jul 14, 2020

View reviewed changes

tombentley added 4 commits July 14, 2020 16:18

KAFKA-10120: Deprecate and replace DescribeLogDirsResult.all() and .v…

dcb214c

…alues() As per KIP-621. Also added some tests in KafkaAdminClientTest

Review comments

05c63fb

Review comments

f2232b7

Review comment

ba2160c

tombentley force-pushed the KAFKA-10120-DescribeLogDirsResult-exposes-internal-classes branch from 33a9413 to ba2160c Compare July 14, 2020 15:21

chia7712 reviewed Jul 14, 2020

View reviewed changes

Review comment

590d9e5

dajac reviewed Jul 14, 2020

View reviewed changes

tombentley added 2 commits July 15, 2020 10:35

Code review

af5224e

Review comment

cad6e63

dongjinleekr suggested changes Jul 15, 2020

View reviewed changes

Review comment

7fb3f12

Review comment

637e7db

dongjinleekr approved these changes Jul 15, 2020

View reviewed changes

dajac approved these changes Jul 16, 2020

View reviewed changes

chia7712 approved these changes Jul 16, 2020

View reviewed changes

mimaison requested changes Jul 23, 2020

View reviewed changes

Review comments

d17d02f

mimaison approved these changes Jul 30, 2020

View reviewed changes

mimaison merged commit 819cd45 into apache:trunk Jul 30, 2020

baluchicken mentioned this pull request Feb 17, 2021

Add support for kafka version 2.7 linkedin/cruise-control#1471

Merged

il-kyun mentioned this pull request Sep 2, 2024

Replace DescribeLogDirsResult.values method with descriptions method linkedin/cruise-control#2190

Open

		case class ReplicaInfo(size: Long, offsetLag: Long, isFuture: Boolean)
		case class LogDirInfo(error: Errors, replicaInfos: Map[TopicPartition, ReplicaInfo])

KAFKA-10120: Deprecate DescribeLogDirsResult.all() and .values() #9007

KAFKA-10120: Deprecate DescribeLogDirsResult.all() and .values() #9007

Conversation

tombentley commented Jul 10, 2020

tombentley commented Jul 10, 2020

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tombentley commented Jul 13, 2020

Choose a reason for hiding this comment

dongjinleekr left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tombentley commented Jul 14, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

dajac left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tombentley commented Jul 15, 2020

dajac left a comment

Choose a reason for hiding this comment

omkreddy commented Jul 23, 2020

mimaison left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tombentley commented Jul 28, 2020

mimaison commented Jul 30, 2020

mimaison commented Jul 30, 2020

mimaison commented Jul 30, 2020

dongjinleekr commented Aug 1, 2020