dengziming commented on a change in pull request #9754:
URL: https://github.com/apache/kafka/pull/9754#discussion_r545026872
##########
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:
@chia7712 Thank you, I removed the version when serialize, and add a
loop when deserialize, Do you think that's right.
----------------------------------------------------------------
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]
