KAFKA-9434: automated protocol for alterReplicaLogDirs by tombentley · Pull Request #8311 · apache/kafka

tombentley · 2020-03-18T11:13:14Z

Change the alterReplicaLogDirs request and response to use the generated protocol messages.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

tombentley · 2020-03-18T11:13:36Z

@mimaison would you be able to review this one?

tombentley · 2020-04-09T12:39:20Z

Rebased for conflict.

tombentley · 2020-04-09T14:58:27Z

Rebased for conflict.

dajac

@tombentley Thanks for the PR. I have made a pass over it and I have left couple of comments.

dajac · 2020-04-09T14:52:57Z

While we are refactoring this, could we add a unit test in KafkaApisTest?

tombentley · 2020-04-21T09:36:07Z

@dajac thanks for the review, I've addressed all your comments.

dajac

@tombentley Thanks for the updated PR. The overall logic looks good to me. I have left a bunch of small comments.

dajac · 2020-04-21T15:31:04Z

nit: Could we rename value to something like alterReplicaLogDirs?

dajac · 2020-04-22T08:23:36Z

Could we add a unit test for this one and the one bellow?

dajac · 2020-04-22T08:24:42Z

Could we add a unit test for this one?

tombentley · 2020-04-22T13:00:39Z

Thanks @dajac, there were some useful comments and good spots there. I've pushed some fixes.

dajac

@tombentley Thanks for the updated PR. It looks good overall. I have left few more small comments and one question.

dajac · 2020-04-23T06:37:53Z

assertNulls shouldn't be here but few lines bellow.

Good catch! Sorry about that.

dajac · 2020-04-23T06:50:32Z

@tombentley This one bugs me. I know that it was already like this in the previous implementation but I wonder if we could improve this while we are here. I am also fine keeping it as it is and improving this latter on.

I find odd that we fail all the non-realised yet futures when we discover an unknown replica in the response. From a client perspective, the behaviour is unpredictable and we leave it with potentially a partial view of the results event though the brokers may have sent back all the requested ones. Wouldn't it be better to just ignore unexpected replica and log a warning message? If a broker would ever do this, the client would continue to work as expected which is probably better. I think that it is a better behavior.

Similarly, we may want to ensure that all futures are done in case a replica would be missing in the response from the broker.

What do you think?

dajac · 2020-04-23T06:57:17Z

Do you mind adding a unit test for this one in AlterReplicaLogDirsRequestTest for the sake of completeness?

tombentley · 2020-04-27T13:29:39Z

Rebased for conflict.

@dajac thanks for the review. I agree about the case of the unexpected partition. The only way it could happen would be a bug in either the client or the broker, but there's no need to force that onto the user to handle. A warning log should be enough for the bug to be reported eventually. So I've made this change.

About the case of the client not getting a response for a partition: I agree that could happen (again, as a bug in the broker or client). It would be a bit harder to detect though; we'd be spending CPU cycles to detect and log a case which we don't think will ever happen in practice. Maybe that's OK since altering log dirs not something done frequently. But it strikes me that there are plenty of other RPCs in the admin client that don't do this kind of second-guessing of the results. If we're prepared to go to that effort here, why not elsewhere? So personally, I think it would OK to omit this check, but I can do it if you really think it should be done.

dajac · 2020-05-04T15:38:02Z

@tombentley Did you consider adding a loop in at the end of the handleResponse that would complete any non-completed future related to the current brokerId with an exception?

// The server should send back a response for every topic. But do a sanity check anyway.
for (Map.Entry<TopicPartitionReplica, KafkaFutureImpl<Void>> entry : futures.entrySet()) {
    TopicPartitionReplica replica = entry.getKey();
    KafkaFutureImpl<Void> future = entry.getValue();
    if (!future.isDone && replica.brokerId() == brokerId)) {
        future.completeExceptionally(new ApiException("The server response did not " +
            "contain a reference to node " + entry.getKey()));
    }
}

You're right. We have few other cases in the Admin client that suffer from the same issue. If we can find an easy way to do it, I would rather prefer doing it in this PR so we can forget about it. I will try to fix some of the others myself.

tombentley · 2020-05-04T17:14:11Z

@dajac yes, fair point. And I notice now that create and delete topics response handlers are already doing this (though the error messages in those cases don't really make much sense). So I've added this with a slightly reworded message.

dajac · 2020-05-05T09:42:26Z

@tombentley thanks for the addition. we could perhaps add a small unit test for the sake of completeness. we already cover all the other cases but this one.

tombentley · 2020-05-05T17:12:48Z

@dajac sure. I added tests for the create and delete topics cases too.

dajac

@tombentley Thanks for the update! LGTM. I just left small cosmetic comments.

dajac · 2020-05-06T08:18:14Z

+                        if (!future.isDone()
+                                && replica.brokerId() == brokerId) {


nit: i would bring the second part on the first line.

dajac · 2020-05-06T08:23:43Z

+    /**
+     * Fail the given future when a response handler expected a result for an entity but no result was present.
+     * @param future The future to fail.
+     * @param message The message to fail the future with


nit: . at the end to stay consistent with previous line.

dajac · 2020-05-06T08:28:31Z

+     * @param future The future to fail.
+     * @param message The message to fail the future with
+     */
+    private void partialResponse(KafkaFutureImpl<?> future, String message) {


I am not entirely convinced by the name. At the end, the method complete a future with the given message. There is nothing specific to handling partial responses. I would rename it to something more generic or simply remove it as we don't gain much with it.

I agree it's not doing much. What value it has is in trying to handle these cases in a consistent way, and being able to more easily discover/reason about the call sites. Maybe something like invalidBrokerResponse() would be a better name? But if you don't like that I'm happy to remove it.

I see your point. I am definitely for reusing logic as much as possible for those cases. What about going an extra step and do something like this then (code not tested)? Just an idea...

public static <K, V> void completeUnrealizedFutures( Map<K, KafkaFutureImpl<V>> futures, Function<K, Boolean> filter, Function<K, String> messageFormatter) { for (Map.Entry<K, KafkaFutureImpl<V>> entry : futures.entrySet()) { K key = entry.getKey(); KafkaFutureImpl<V> future = entry.getValue(); if (!future.isDone() && filter.apply(key)) { future.completeExceptionally(new ApiException(messageFormatter.apply(key))); } } } public static <K, V> void completeUnrealizedFutures( Map<K, KafkaFutureImpl<V>> futures, unction<K, String> messageFormatter) { completeUnrealizedFutures(futures, (key) -> true, messageFormatter); }

It would allow the following:

completeUnrealizedFutures(topicFutures, replica -> replica.brokerId() == brokerId, topic -> "The response from broker " + brokerId + " did not contain a result for replica " + replica); completeUnrealizedFutures(topicFutures, topic -> "The server response did not contain a reference to node " + topic);

That works for me, especially if this is something you're going to make more use of for the other response handlers in the admin client.

dajac

@tombentley Thanks for the update. The PR LGTM! I just left few cosmetic comments.

dajac · 2020-05-08T11:25:00Z

-                }
+                completeUnrealizedFutures(topicFutures,
+                    topic -> "The controller response " +
+                            "did not contain a result for topic " + topic);


nit: we can bring this back on the first line.

dajac · 2020-05-08T11:25:05Z

-                }
+                completeUnrealizedFutures(topicFutures,
+                    topic -> "The controller response " +
+                            "did not contain a result for topic " + topic);


nit: we can bring this back on the first line.

dajac · 2020-05-08T11:25:48Z

-                            future.completeExceptionally(error.exception());
+                    for (AlterReplicaLogDirTopicResult topicResult: response.data().results()) {
+                        for (AlterReplicaLogDirPartitionResult partitionResult: topicResult.partitions()) {
+                            TopicPartitionReplica replica = new TopicPartitionReplica(topicResult.topicName(), partitionResult.partitionIndex(), brokerId);


nit: We could break this long line.

tombentley · 2020-05-11T10:18:39Z

@dajac I fixed those, and also changed the completeUnrealizedFutures() method to take a Stream, which is a little nicer since it allows the caller to do the filtering, thus removing the need for the overloaded version.

dajac · 2020-05-11T15:05:55Z

@tombentley thanks!

hachikuji · 2020-05-20T01:27:39Z

ok to test

mimaison · 2020-05-28T10:13:46Z

retest this please

mimaison · 2020-06-04T09:31:03Z

+            try {
+                topicsResult.values().get("myTopic2").get();
+                fail("Expected an exception.");
+            } catch (ExecutionException e) {


We can use assertThrows() here. Same below

I don't think we can, at least not if we want to check that topicsResult.values().get("myTopic2").get() throws an ExecutionException wrapping an ApiException. assertThrows() would only let us assert the outer exception type.

Sorry, I meant TestUtils.assertFutureThrows()

Ah, thanks! Now fixed.

mimaison · 2020-06-04T09:37:03Z

+                                    topicResult.topicName(), partitionResult.partitionIndex(), brokerId);
+                            KafkaFutureImpl<Void> future = futures.get(replica);
+                            if (future == null) {
+                                log.warn("The partition {} in the response from broker {}} is not in the request",


There's an extra }

mimaison · 2020-06-04T09:41:04Z

Thanks @tombentley for the PR. It looks good. Happy to merge once the last 2 comments are addressed

mimaison · 2020-06-04T13:09:16Z

retest this please

tombentley · 2020-06-04T13:26:40Z

@mimaison done.

mimaison · 2020-06-04T14:07:26Z

retest this please

* 'trunk' of github.com:apache/kafka: (46 commits) MINOR: improve code encapsulation between StreamThread and TaskManager (apache#8819) Fixing KAFKA-10094 (apache#8797) KAFKA-9851: Revoking Connect tasks due to connectivity issues should also clear the running assignment (apache#8804) KAFKA-9840; Skip End Offset validation when the leader epoch is not reliable (apache#8486) HOT_FIX: Update javadoc since imports added (apache#8817) KAFKA-8011: Fix flaky RegexSourceIntegrationTest (apache#8799) KAFKA-9570: Define SSL configs in all worker config classes, not just distributed (apache#8135) KAFKA-10111: Make SinkTaskContext.errantRecordReporter() a default method (apache#8814) KAFKA-10110: Corrected potential NPE when null label value added to KafkaMetricsContext (apache#8811) MINOR: Change the order that Connect calls `config()` and `validate()` to avoid validating if the required ConfigDef is null (apache#8810) MINOR: fix backwards incompatibility in JmxReporter introduced by KIP-606 MINOR: Fix javadoc warnings (apache#8809) KAFKA-9441: Improve Kafka Streams task management (apache#8776) fix the broken links of streams javadoc (apache#8789) KAFKA-10040; Make computing the PreferredReplicaImbalanceCount metric more efficient (apache#8724) KAFKA-10066: TestOutputTopic should pass record headers into deserializers (apache#8759) MINOR: Add explanation for disabling forwarding from value transformers (apache#8771) KAFKA-10033: Throw UnknownTopicOrPartitionException if altering configs of non-existing topic KAFKA-9434: automated protocol for alterReplicaLogDirs (apache#8311) KAFKA-9313: Set `use_all_dns_ips` as the new default for `client.dns.lookup` (KIP-602) (apache#8644) ...

tombentley force-pushed the KAFKA-9434-AlterReplicaLogDirs-automatedprotocol branch from e154815 to 7658599 Compare March 20, 2020 16:07

tombentley force-pushed the KAFKA-9434-AlterReplicaLogDirs-automatedprotocol branch from 7658599 to 3c32ac4 Compare April 9, 2020 12:36

tombentley force-pushed the KAFKA-9434-AlterReplicaLogDirs-automatedprotocol branch from 3c32ac4 to b1584aa Compare April 9, 2020 14:54

dajac reviewed Apr 9, 2020

View reviewed changes

tombentley force-pushed the KAFKA-9434-AlterReplicaLogDirs-automatedprotocol branch from a6d1e17 to 4cb5e42 Compare April 21, 2020 09:35

dajac reviewed Apr 22, 2020

View reviewed changes

dajac reviewed Apr 23, 2020

View reviewed changes

tombentley added 13 commits April 27, 2020 14:08

wip, need to fix KafkaApis/ReplicaManager

c218317

wip

9a33890

Minor improvements

21f7375

Review comments

a427748

review comments

f9069aa

Add test for KafkaAdminClient.alterReplicaLogDirs

d3fd28c

Add test for KafkaAdminClient.alterReplicaLogDirs

ff5e457

Add test for AlterLogDirs RPCs to RequestResponseTests

a4bcdc0

Suggested code by @dajac

e889378

Add KafkaApisTest.testAlterReplicaLogDirs()

01a2980

Fix test

eaf80dc

review comments

0fc3bc3

Review comments; change handling of unexpected responses

1dffaa9

tombentley force-pushed the KAFKA-9434-AlterReplicaLogDirs-automatedprotocol branch from e323868 to 1dffaa9 Compare April 27, 2020 13:09

Review comments 2

c82455f

Exception on partial response

4c5d1e6

tombentley added 2 commits May 5, 2020 17:09

Add extra test

e0fc9bf

Two more tests

103a123

dajac reviewed May 6, 2020

View reviewed changes

tombentley added 2 commits May 6, 2020 10:02

Code review

0ebdc39

completeUnrealizedFutures

3d1aa65

dajac approved these changes May 8, 2020

View reviewed changes

Review comments + improvement

6a794a6

mimaison mentioned this pull request Jun 3, 2020

KAFKA-9627: Replace ListOffset request/response with automated protocol #8295

Merged

3 tasks

mimaison requested changes Jun 4, 2020

View reviewed changes

mimaison mentioned this pull request Jun 4, 2020

KAFKA-9432 automated protocol for DescribeConfigs #8312

Merged

tombentley added 2 commits June 4, 2020 11:45

review comment

7b6f3c0

TestUtils.assertFutureThrows

66f796c

mimaison approved these changes Jun 4, 2020

View reviewed changes

mimaison requested changes Jun 4, 2020

View reviewed changes

Comment thread clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java Outdated

Comment thread clients/src/test/java/org/apache/kafka/clients/admin/KafkaAdminClientTest.java Outdated

More TestUtils.assertFutureThrows

2af415b

mimaison approved these changes Jun 4, 2020

View reviewed changes

mimaison merged commit 78e8a49 into apache:trunk Jun 4, 2020

Uh oh!

Conversation

tombentley commented Mar 18, 2020

Committer Checklist (excluded from commit message)

Uh oh!

tombentley commented Mar 18, 2020

Uh oh!

tombentley commented Apr 9, 2020

Uh oh!

tombentley commented Apr 9, 2020

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

tombentley commented Apr 21, 2020

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tombentley commented Apr 22, 2020

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

tombentley commented Apr 27, 2020

Uh oh!

dajac commented May 4, 2020

Uh oh!

tombentley commented May 4, 2020

Uh oh!

dajac commented May 5, 2020

Uh oh!

tombentley commented May 5, 2020

Uh oh!

dajac left a comment

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!