You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by GitBox <gi...@apache.org> on 2020/10/26 20:42:19 UTC
[GitHub] [kafka] dajac opened a new pull request #9506: KAFKA-10647; Only serialize owned partition when consumer protocol version >= 0
dajac opened a new pull request #9506:
URL: https://github.com/apache/kafka/pull/9506
A regression got introduced by https://github.com/apache/kafka/pull/8897. The owned partition field must be ignored for version < 1 otherwise the serialization fails with an unsupported version exception.
### Committer Checklist (excluded from commit message)
- [ ] Verify design and implementation
- [ ] Verify test coverage and CI build status
- [ ] Verify documentation (including upgrade notes)
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] hachikuji commented on a change in pull request #9506: KAFKA-10647; Only serialize owned partitions when consumer protocol version >= 1
Posted by GitBox <gi...@apache.org>.
hachikuji commented on a change in pull request #9506:
URL: https://github.com/apache/kafka/pull/9506#discussion_r512344373
##########
File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
##########
@@ -46,6 +46,30 @@
private final TopicPartition tp2 = new TopicPartition("bar", 2);
private final Optional<String> groupInstanceId = Optional.of("instance.id");
+ @Test
+ public void serializeDeserializeSubscriptionAllVersions() {
+ List<TopicPartition> ownedPartitions = Arrays.asList(
+ new TopicPartition("foo", 0),
+ new TopicPartition("bar", 0));
+ Subscription subscription = new Subscription(Arrays.asList("foo", "bar"),
+ ByteBuffer.wrap("hello".getBytes()), ownedPartitions);
+
+ for (short version = ConsumerProtocolSubscription.LOWEST_SUPPORTED_VERSION; version <= ConsumerProtocolSubscription.HIGHEST_SUPPORTED_VERSION; version++) {
+ ByteBuffer buffer = ConsumerProtocol.serializeSubscription(subscription, version);
+ Subscription parsedSubscription = ConsumerProtocol.deserializeSubscription(buffer);
+
+ assertEquals(subscription.topics(), parsedSubscription.topics());
+ assertEquals(subscription.userData(), parsedSubscription.userData());
+ assertFalse(parsedSubscription.groupInstanceId().isPresent());
+
+ if (version >= 1) {
+ assertEquals(toSet(subscription.ownedPartitions()), toSet(parsedSubscription.ownedPartitions()));
+ } else {
+ assertTrue(parsedSubscription.ownedPartitions().isEmpty());
Review comment:
nit: A better general pattern is to use `assertEquals` comparing against empty list. Then if the assertion fails, the message will show what was in the collection.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on pull request #9506: KAFKA-10647; Only serialize owned partitions when consumer protocol version >= 1
Posted by GitBox <gi...@apache.org>.
dajac commented on pull request #9506:
URL: https://github.com/apache/kafka/pull/9506#issuecomment-717132358
@hachikuji @chia7712 Thanks for your review! Merging to trunk and 2.7.
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac merged pull request #9506: KAFKA-10647; Only serialize owned partitions when consumer protocol version >= 1
Posted by GitBox <gi...@apache.org>.
dajac merged pull request #9506:
URL: https://github.com/apache/kafka/pull/9506
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] dajac commented on a change in pull request #9506: KAFKA-10647; Only serialize owned partitions when consumer protocol version >= 1
Posted by GitBox <gi...@apache.org>.
dajac commented on a change in pull request #9506:
URL: https://github.com/apache/kafka/pull/9506#discussion_r512458078
##########
File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
##########
@@ -46,6 +46,30 @@
private final TopicPartition tp2 = new TopicPartition("bar", 2);
private final Optional<String> groupInstanceId = Optional.of("instance.id");
+ @Test
+ public void serializeDeserializeSubscriptionAllVersions() {
+ List<TopicPartition> ownedPartitions = Arrays.asList(
+ new TopicPartition("foo", 0),
+ new TopicPartition("bar", 0));
+ Subscription subscription = new Subscription(Arrays.asList("foo", "bar"),
+ ByteBuffer.wrap("hello".getBytes()), ownedPartitions);
+
+ for (short version = ConsumerProtocolSubscription.LOWEST_SUPPORTED_VERSION; version <= ConsumerProtocolSubscription.HIGHEST_SUPPORTED_VERSION; version++) {
+ ByteBuffer buffer = ConsumerProtocol.serializeSubscription(subscription, version);
+ Subscription parsedSubscription = ConsumerProtocol.deserializeSubscription(buffer);
+
+ assertEquals(subscription.topics(), parsedSubscription.topics());
+ assertEquals(subscription.userData(), parsedSubscription.userData());
+ assertFalse(parsedSubscription.groupInstanceId().isPresent());
+
+ if (version >= 1) {
+ assertEquals(toSet(subscription.ownedPartitions()), toSet(parsedSubscription.ownedPartitions()));
+ } else {
+ assertTrue(parsedSubscription.ownedPartitions().isEmpty());
Review comment:
Good point.
##########
File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
##########
@@ -137,6 +161,19 @@ public void deserializeFutureSubscriptionVersion() {
assertEquals(groupInstanceId, subscription.groupInstanceId());
}
+ @Test
+ public void serializeDeserializeAssignmentAllVersions() {
+ List<TopicPartition> partitions = Arrays.asList(tp1, tp2);
+ Assignment assignment = new Assignment(partitions, ByteBuffer.wrap("hello".getBytes()));
+
+ for (short version = ConsumerProtocolAssignment.LOWEST_SUPPORTED_VERSION; version <= ConsumerProtocolAssignment.HIGHEST_SUPPORTED_VERSION; version++) {
+ ByteBuffer buffer = ConsumerProtocol.serializeAssignment(assignment, version);
+ Assignment parsedAssignment = ConsumerProtocol.deserializeAssignment(buffer);
+ assertEquals(toSet(partitions), toSet(parsedAssignment.partitions()));
+ assertEquals(parsedAssignment.userData(), parsedAssignment.userData());
Review comment:
Good catch!
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] chia7712 commented on a change in pull request #9506: KAFKA-10647; Only serialize owned partitions when consumer protocol version >= 1
Posted by GitBox <gi...@apache.org>.
chia7712 commented on a change in pull request #9506:
URL: https://github.com/apache/kafka/pull/9506#discussion_r512397607
##########
File path: clients/src/test/java/org/apache/kafka/clients/consumer/internals/ConsumerProtocolTest.java
##########
@@ -137,6 +161,19 @@ public void deserializeFutureSubscriptionVersion() {
assertEquals(groupInstanceId, subscription.groupInstanceId());
}
+ @Test
+ public void serializeDeserializeAssignmentAllVersions() {
+ List<TopicPartition> partitions = Arrays.asList(tp1, tp2);
+ Assignment assignment = new Assignment(partitions, ByteBuffer.wrap("hello".getBytes()));
+
+ for (short version = ConsumerProtocolAssignment.LOWEST_SUPPORTED_VERSION; version <= ConsumerProtocolAssignment.HIGHEST_SUPPORTED_VERSION; version++) {
+ ByteBuffer buffer = ConsumerProtocol.serializeAssignment(assignment, version);
+ Assignment parsedAssignment = ConsumerProtocol.deserializeAssignment(buffer);
+ assertEquals(toSet(partitions), toSet(parsedAssignment.partitions()));
+ assertEquals(parsedAssignment.userData(), parsedAssignment.userData());
Review comment:
Does it compare to incorrect object (maybe ```assignment.userData()```)?
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
users@infra.apache.org