chia7712 commented on a change in pull request #9754:
URL: https://github.com/apache/kafka/pull/9754#discussion_r544131964
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
##########
@@ -222,51 +204,50 @@ protected MemberData memberData(Subscription
subscription) {
return deserializeTopicPartitionAssignment(userData);
}
- // visible for testing
static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData)
{
- Struct struct = new Struct(STICKY_ASSIGNOR_USER_DATA_V1);
- List<Struct> topicAssignments = new ArrayList<>();
+ return serializeTopicPartitionAssignment(memberData,
StickyAssignorUserData.HIGHEST_SUPPORTED_VERSION);
+ }
+
+ // visible for testing
+ static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData,
short version) {
+
+ List<StickyAssignorUserData.TopicPartition> topicAssignments = new
ArrayList<>();
for (Map.Entry<String, List<Integer>> topicEntry :
CollectionUtils.groupPartitionsByTopic(memberData.partitions).entrySet()) {
- Struct topicAssignment = new Struct(TOPIC_ASSIGNMENT);
- topicAssignment.set(TOPIC_KEY_NAME, topicEntry.getKey());
- topicAssignment.set(PARTITIONS_KEY_NAME,
topicEntry.getValue().toArray());
- topicAssignments.add(topicAssignment);
+ StickyAssignorUserData.TopicPartition topicPartition = new
StickyAssignorUserData.TopicPartition()
+ .setTopic(topicEntry.getKey())
+ .setPartitions(topicEntry.getValue());
+ topicAssignments.add(topicPartition);
+ }
+ StickyAssignorUserData data = new StickyAssignorUserData()
+ .setPreviousAssignment(topicAssignments);
+ if (version >= 1) {
Review comment:
How about using ```"ignorable": true``` to eliminate this if-else?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
##########
@@ -222,51 +204,50 @@ protected MemberData memberData(Subscription
subscription) {
return deserializeTopicPartitionAssignment(userData);
}
- // visible for testing
static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData)
{
- Struct struct = new Struct(STICKY_ASSIGNOR_USER_DATA_V1);
- List<Struct> topicAssignments = new ArrayList<>();
+ return serializeTopicPartitionAssignment(memberData,
StickyAssignorUserData.HIGHEST_SUPPORTED_VERSION);
+ }
+
+ // visible for testing
+ static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData,
short version) {
+
+ List<StickyAssignorUserData.TopicPartition> topicAssignments = new
ArrayList<>();
for (Map.Entry<String, List<Integer>> topicEntry :
CollectionUtils.groupPartitionsByTopic(memberData.partitions).entrySet()) {
Review comment:
Why we need
```CollectionUtils.groupPartitionsByTopic(memberData.partitions)``` here ?
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
##########
@@ -222,51 +204,50 @@ protected MemberData memberData(Subscription
subscription) {
return deserializeTopicPartitionAssignment(userData);
}
- // visible for testing
static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData)
{
- Struct struct = new Struct(STICKY_ASSIGNOR_USER_DATA_V1);
- List<Struct> topicAssignments = new ArrayList<>();
+ return serializeTopicPartitionAssignment(memberData,
StickyAssignorUserData.HIGHEST_SUPPORTED_VERSION);
+ }
+
+ // visible for testing
+ static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData,
short version) {
+
+ List<StickyAssignorUserData.TopicPartition> topicAssignments = new
ArrayList<>();
for (Map.Entry<String, List<Integer>> topicEntry :
CollectionUtils.groupPartitionsByTopic(memberData.partitions).entrySet()) {
- Struct topicAssignment = new Struct(TOPIC_ASSIGNMENT);
- topicAssignment.set(TOPIC_KEY_NAME, topicEntry.getKey());
- topicAssignment.set(PARTITIONS_KEY_NAME,
topicEntry.getValue().toArray());
- topicAssignments.add(topicAssignment);
+ StickyAssignorUserData.TopicPartition topicPartition = new
StickyAssignorUserData.TopicPartition()
+ .setTopic(topicEntry.getKey())
+ .setPartitions(topicEntry.getValue());
+ topicAssignments.add(topicPartition);
+ }
+ StickyAssignorUserData data = new StickyAssignorUserData()
+ .setPreviousAssignment(topicAssignments);
+ if (version >= 1) {
+ memberData.generation.ifPresent(data::setGeneration);
}
- struct.set(TOPIC_PARTITIONS_KEY_NAME, topicAssignments.toArray());
- if (memberData.generation.isPresent())
- struct.set(GENERATION_KEY_NAME, memberData.generation.get());
- ByteBuffer buffer =
ByteBuffer.allocate(STICKY_ASSIGNOR_USER_DATA_V1.sizeOf(struct));
- STICKY_ASSIGNOR_USER_DATA_V1.write(buffer, struct);
- buffer.flip();
- return buffer;
+ return MessageUtil.toVersionPrefixedByteBuffer(version, data);
}
- private static MemberData deserializeTopicPartitionAssignment(ByteBuffer
buffer) {
- Struct struct;
- ByteBuffer copy = buffer.duplicate();
+ private static MemberData deserializeTopicPartitionAssignment(ByteBuffer
buffer, short version) {
+ StickyAssignorUserData data;
try {
- struct = STICKY_ASSIGNOR_USER_DATA_V1.read(buffer);
+ data = new StickyAssignorUserData(new ByteBufferAccessor(buffer),
version);
Review comment:
It seems to me previous serialization does not carry version field so we
have to use this ugly code to handle different version.
##########
File path:
clients/src/main/java/org/apache/kafka/clients/consumer/StickyAssignor.java
##########
@@ -222,51 +204,50 @@ protected MemberData memberData(Subscription
subscription) {
return deserializeTopicPartitionAssignment(userData);
}
- // visible for testing
static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData)
{
- Struct struct = new Struct(STICKY_ASSIGNOR_USER_DATA_V1);
- List<Struct> topicAssignments = new ArrayList<>();
+ return serializeTopicPartitionAssignment(memberData,
StickyAssignorUserData.HIGHEST_SUPPORTED_VERSION);
+ }
+
+ // visible for testing
+ static ByteBuffer serializeTopicPartitionAssignment(MemberData memberData,
short version) {
+
+ List<StickyAssignorUserData.TopicPartition> topicAssignments = new
ArrayList<>();
for (Map.Entry<String, List<Integer>> topicEntry :
CollectionUtils.groupPartitionsByTopic(memberData.partitions).entrySet()) {
- Struct topicAssignment = new Struct(TOPIC_ASSIGNMENT);
- topicAssignment.set(TOPIC_KEY_NAME, topicEntry.getKey());
- topicAssignment.set(PARTITIONS_KEY_NAME,
topicEntry.getValue().toArray());
- topicAssignments.add(topicAssignment);
+ StickyAssignorUserData.TopicPartition topicPartition = new
StickyAssignorUserData.TopicPartition()
+ .setTopic(topicEntry.getKey())
+ .setPartitions(topicEntry.getValue());
+ topicAssignments.add(topicPartition);
+ }
+ StickyAssignorUserData data = new StickyAssignorUserData()
+ .setPreviousAssignment(topicAssignments);
+ if (version >= 1) {
+ memberData.generation.ifPresent(data::setGeneration);
}
- struct.set(TOPIC_PARTITIONS_KEY_NAME, topicAssignments.toArray());
- if (memberData.generation.isPresent())
- struct.set(GENERATION_KEY_NAME, memberData.generation.get());
- ByteBuffer buffer =
ByteBuffer.allocate(STICKY_ASSIGNOR_USER_DATA_V1.sizeOf(struct));
- STICKY_ASSIGNOR_USER_DATA_V1.write(buffer, struct);
- buffer.flip();
- return buffer;
+ return MessageUtil.toVersionPrefixedByteBuffer(version, data);
Review comment:
Does previous serialization have version field?
----------------------------------------------------------------
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:
[email protected]
