dengziming commented on a change in pull request #9754:
URL: https://github.com/apache/kafka/pull/9754#discussion_r546640844
##########
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:
Do you think we need flexible version support here? @chia7712
----------------------------------------------------------------
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]
