KAFKA-4635: Client Compatibility follow-ups #2414

cmccabe · 2017-01-20T23:31:40Z

No description provided.

asfbot · 2017-01-20T23:34:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1078/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-20T23:34:13Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1080/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-20T23:36:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1078/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-21T00:35:56Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1081/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-21T00:47:05Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1079/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-21T02:22:53Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1079/
Test PASSed (JDK 7 and Scala 2.10).

ijuma · 2017-01-25T10:23:51Z

clients/src/main/java/org/apache/kafka/common/errors/OutdatedBrokerException.java

@@ -13,18 +13,18 @@
 package org.apache.kafka.common.errors;

 /**
- * Indicates that a request cannot be completed because an obsolete broker
+ * Indicates that a request cannot be completed because an outdated broker
 * does not support the required functionality.


I was thinking about this and it seems that we can also get this error if the client requires a protocol version that is too old for the broker. If that's correct, then maybe the name could be misleading.

That is a good point. We should have a separate exception for when the client is outdated.

asfbot · 2017-01-25T23:44:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1203/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-25T23:53:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1201/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-26T00:16:24Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1201/
Test PASSed (JDK 7 and Scala 2.10).

ijuma · 2017-01-26T10:58:13Z

One additional thing: it would be good to mention in NodeApiVersions's class Javadoc that it's an internal class.

ijuma · 2017-01-26T14:52:07Z

I stole the upgrade notes bit and added it to #2445

ijuma

Thanks for the PR, left a few minor comments.

ijuma · 2017-01-26T14:53:09Z

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

+            if (header.apiVersion() == latestClientVersion) {
+                log.trace("Sending {} to node {}.", request, nodeId);
+            } else {
+                log.debug("Using older server API v{} to send {} to node {}.",


@junrao, is this what you had in mind regarding logging of cases where we are downgrading the protocol version?

Yes, this looks good to me.

ijuma · 2017-01-26T14:56:05Z

docs/upgrade.html

@@ -18,6 +18,9 @@
 Users upgrading from versions prior to 0.10.1.x should follow the upgrade guide <a href="#upgrade_10_1">here</a>. Users upgrading from 0.10.1.0
 can upgrade the brokers one at a time: shut down the broker, update the code, and restart it.

+<p>Starting with version 0.10.2, the Kafka Java client is compatible with older brokers.  Version 0.10.2 clients can talk to version 0.10.0 or newer brokers.
+However, if your brokers are older than 0.10.0, you must upgrade the broker before rolling out new clients.


Thinking about it, we should probably mention in the class javadoc of KafkaConsumer and KafkaProducer the behaviour with regards to older brokers.

Yeah.

P.S. I removed the release notes from this patch, since it is being moved to KAFKA-4578.

…r OffsetAndTimestamp

…der broker

* Convert OutdatedBrokerException -> UnsupportedVersionException. * NodeApiVersions: distinguish versions which are too old from those which are too new

cmccabe · 2017-01-26T18:24:43Z

Thanks for the reviews, all. As requested, I removed the release notes from this patch, since it is being moved to KAFKA-4578. Let me rebase to get rid of these conflicts....

asfbot · 2017-01-26T19:25:43Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1263/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-26T19:57:44Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1261/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-26T19:57:51Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1261/
Test PASSed (JDK 7 and Scala 2.10).

hachikuji · 2017-01-27T01:16:39Z

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

@@ -88,7 +87,7 @@ public Builder setTargetTimes(Map<TopicPartition, Long> partitionTimestamps) {
        public ListOffsetRequest build() {
            short version = version();
            if (version < minVersion) {
-                throw new ObsoleteBrokerException("The broker is too old to send this request.");
+                throw new UnsupportedVersionException("The broker is too old to understand this request.");


I feel bad for the broker. It's not too old; it's just fallen behind the times a bit.

By the way, could we add some more info to this message, such as which api is used, which version is needed and which is supported by the broker. Same for the other below.

Good point. I added information about the API being called and the version sent (versus required) to the exception.

For the exception below, it should not actually be reachable unless there is a programming error, so it is better off as a RuntimeException rather than an UnsupportedVersionException.

…ionException

hachikuji · 2017-01-27T01:35:10Z

clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndTimestamp.java

 */
 public final class OffsetAndTimestamp {
+    // The timestamp should never be negative, unless it is invalid.  This could happen when


This comment still makes me uncomfortable. For a public-facing class, it would be nice to have surer guarantee about the class invariants (e.g. by checking that the timestamp is not negative in the constructor). I wonder if there is a different way we can represent the sentinel value internally in Fetcher.handleListOffsetResponse. For example maybe we just need a static instance that we can use in Fetcher whenever we need the sentinel and which we can check for using ==?

I agree that allowing negatives here feels uncomfortable. Unfortunately, it reflects the fundamental fact that old servers are not sending back timestamps, and new servers are. I tried to create a totally separate code path for the old servers, but it was simply too much code duplication. Also unfortunately, because we need to read the offset from the structures, we cannot use static instances.

Probably the best way around this problem is to stop using the public type internally. Then we can have an internal type where the timestamp is optional, and a public type where it is not. This also makes it less likely that we'll break the public API by making an internal change, so it shoudl be a good thing overall. The downside is that copying will be needed, but I don't think the amount of copying is significant for this use-case.

hachikuji · 2017-01-27T01:41:21Z

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -290,19 +289,16 @@ private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long
            // information itself.  It is also the case when discoverBrokerVersions is set to false.
            if (versionInfo == null) {
                if ((!discoverBrokerVersions) && (log.isTraceEnabled()))
-                    log.trace("No version information found when sending message of type {} to node {}",
-                            clientRequest.apiKey(), nodeId);
+                    log.trace("No version information found when sending message of type {} to node {}.  " +


nit: We use a single space after a period throughout the rest of the code. It would be good to stay consistent.

asfbot · 2017-01-27T02:33:35Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1281/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-27T19:42:45Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1297/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-27T19:42:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1297/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-27T19:42:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1299/
Test FAILed (JDK 8 and Scala 2.11).

cmccabe · 2017-01-27T19:50:16Z

@ijuma: I added a @throws clause to KafkaProducer#offsetsForTimes.

asfbot · 2017-01-27T19:52:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1300/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-27T19:53:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1298/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-27T19:53:45Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1298/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-27T20:51:01Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1301/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-27T20:52:37Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1299/
Test PASSed (JDK 7 and Scala 2.10).

asfbot · 2017-01-27T21:22:08Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1299/
Test PASSed (JDK 8 and Scala 2.12).

ijuma · 2017-01-27T20:53:20Z

clients/src/main/java/org/apache/kafka/clients/consumer/OffsetAndTimestamp.java

        this.timestamp = timestamp;
+        assert(this.timestamp >= 0);


We typically avoid Java language asserts because they are disabled by default.

ijuma

Thanks for the updates, 2 minor comments, LGTM otherwise.

ijuma · 2017-01-27T22:45:55Z

clients/src/main/java/org/apache/kafka/clients/NetworkClient.java

@@ -290,19 +289,16 @@ private void doSend(ClientRequest clientRequest, boolean isInternalRequest, long
            // information itself.  It is also the case when discoverBrokerVersions is set to false.
            if (versionInfo == null) {
                if ((!discoverBrokerVersions) && (log.isTraceEnabled()))
-                    log.trace("No version information found when sending message of type {} to node {}",
-                            clientRequest.apiKey(), nodeId);
+                    log.trace("No version information found when sending message of type {} to node {}. " +


Is this log message correct? It seems to imply that we looked for the version information and didn't find it. But it seems like the feature is simply disabled due to discoverBrokerVersions = false?

Good catch! The sense of the if statement is inverted-- let me fix that.

ijuma · 2017-01-27T22:48:38Z

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

@@ -60,7 +60,7 @@ public boolean isAllTopicPartitions() {
        @Override
        public OffsetFetchRequest build() {
            if (isAllTopicPartitions() && version() < 2)
-                throw new ObsoleteBrokerException("The broker is too old to send this request.");
+                throw new UnsupportedVersionException("The broker is too old to understand this request.");


Do we want to include version in this message?

asfbot · 2017-01-27T23:14:17Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1307/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-27T23:15:45Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1305/
Test PASSed (JDK 8 and Scala 2.12).

asfbot · 2017-01-27T23:17:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1305/
Test PASSed (JDK 7 and Scala 2.10).

hachikuji

LGTM. Thanks for the fixes! One tiny nit which can be fixed when merging.

hachikuji · 2017-01-28T00:21:16Z

clients/src/main/java/org/apache/kafka/clients/NodeApiVersions.java

-        }
-        return usableVersion;
+        if (usableVersion == API_NOT_ON_NODE)
+            throw new UnsupportedVersionException("The remote node does not support " + apiKey);


nit: "node" is a bit generic for a user message. Maybe it should be "broker"?

I guess the idea here is that NodeApiVersions is intended to handle arbitrary nodes, not just brokers. The idea is that it should be able to be used server-side eventually. Maybe "peer" or "remote" would be a better name, but then the class should be renamed to PeerApiVersions / RemoteApiVersions, and that seemed like a big change for a cleanup.

In Kafka, brokers are still called brokers when the communication is between brokers (the inter.broker.protocol.version config, for example). So, it would not look out of place although I am aware that other projects use different wording, as you said.

The thought was that we would use it when communicating with clients as well, not just brokers.

I can change it for now if it seems too vague, though.

I changed the message to use broker for now. It's true that it looks a bit out of place given the class name, but the class is internal and we can rename it later. The message may be seen by users and it's good to make it as clear as possible.

ijuma · 2017-01-28T00:57:13Z

I'll merge this to trunk and 0.10.2 once Jenkins gives the green light.

asfbot · 2017-01-28T00:58:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/1308/
Test PASSed (JDK 7 and Scala 2.10).

Author: Colin P. Mccabe <cmccabe@confluent.io> Reviewers: Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk> Closes #2414 from cmccabe/KAFKA-4635 (cherry picked from commit 8827a5b) Signed-off-by: Ismael Juma <ismael@juma.me.uk>

asfbot · 2017-01-28T01:47:54Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/1310/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-28T01:50:10Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/1308/
Test PASSed (JDK 8 and Scala 2.12).

Author: Colin P. Mccabe <cmccabe@confluent.io> Reviewers: Jason Gustafson <jason@confluent.io>, Ismael Juma <ismael@juma.me.uk> Closes apache#2414 from cmccabe/KAFKA-4635

ijuma reviewed Jan 25, 2017

View reviewed changes

cmccabe force-pushed the KAFKA-4635 branch from 9554821 to 464f89d Compare January 25, 2017 22:53

ijuma reviewed Jan 26, 2017

View reviewed changes

cmccabe added 5 commits January 26, 2017 09:19

ObsoleteBrokerException -> OutdatedBrokerException, revise comment fo…

260c091

…r OffsetAndTimestamp

Add debug log message when downgrading the message protocol for an ol…

b650419

…der broker

NodeApiVersionsTest: add more tests

b1cc12e

NodeApiVersionsTest: fix checkstyle

90b2e24

Use UnsupportedVersionException

e4d1567

* Convert OutdatedBrokerException -> UnsupportedVersionException. * NodeApiVersions: distinguish versions which are too old from those which are too new

NodeApiVersions: add "internal" JavaDoc

7c57751

cmccabe force-pushed the KAFKA-4635 branch from 464f89d to 7c57751 Compare January 26, 2017 18:36

hachikuji reviewed Jan 27, 2017

View reviewed changes

ListOffsetRequest#Builder#build: add more informative UnsupportedVers…

cac5c50

…ionException

hachikuji reviewed Jan 27, 2017

View reviewed changes

cmccabe added 2 commits January 27, 2017 11:49

OffsetAndTimestamp: fix checkstyle

16ec282

KafkaConsumer#offsetsForTimes: add @throws UnsupportedVersionException

ce2956f

Fetcher.java: fix NPE if entry#getValue is null

bcede0f

ijuma reviewed Jan 27, 2017

View reviewed changes

Add JavaDoc notes to KafkaProducer and KafkaConsumer

c4d86db

ijuma approved these changes Jan 27, 2017

View reviewed changes

cmccabe added 2 commits January 27, 2017 15:35

OffsetFetchRequest: add version information to exception

2af4a34

NetworkClient: fix issue with log message

ed06310

hachikuji approved these changes Jan 28, 2017

View reviewed changes

asfgit closed this in 8827a5b Jan 28, 2017

cmccabe deleted the KAFKA-4635 branch May 20, 2019 18:34

KAFKA-4635: Client Compatibility follow-ups #2414

KAFKA-4635: Client Compatibility follow-ups #2414

Conversation

cmccabe commented Jan 20, 2017

asfbot commented Jan 20, 2017

asfbot commented Jan 20, 2017

asfbot commented Jan 20, 2017

asfbot commented Jan 21, 2017

asfbot commented Jan 21, 2017

asfbot commented Jan 21, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 25, 2017

asfbot commented Jan 25, 2017

asfbot commented Jan 26, 2017

ijuma commented Jan 26, 2017

ijuma commented Jan 26, 2017

ijuma 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

cmccabe commented Jan 26, 2017

asfbot commented Jan 26, 2017

asfbot commented Jan 26, 2017

asfbot commented Jan 26, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

hachikuji Jan 27, 2017 • edited Loading

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

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

cmccabe commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

Choose a reason for hiding this comment

ijuma 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

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

asfbot commented Jan 27, 2017

hachikuji 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

ijuma commented Jan 28, 2017

asfbot commented Jan 28, 2017

asfbot commented Jan 28, 2017

asfbot commented Jan 28, 2017

hachikuji Jan 27, 2017 •

edited

Loading