@Test public void deserializeNewAssignmentVersion() { short version = 100; Schema assignmentSchemaV100 = new Schema( new Field(ConsumerProtocol.TOPIC_PARTITIONS_KEY_NAME, new ArrayOf(ConsumerProtocol.TOPIC_ASSIGNMENT_V0)), new Field(ConsumerProtocol.USER_DATA_KEY_NAME, Type.BYTES), new Field("foo", Type.STRING)); Struct assignmentV100 = new Struct(assignmentSchemaV100); assignmentV100.set(ConsumerProtocol.TOPIC_PARTITIONS_KEY_NAME, new Object[]{new Struct(ConsumerProtocol.TOPIC_ASSIGNMENT_V0) .set(ConsumerProtocol.TOPIC_KEY_NAME, "foo") .set(ConsumerProtocol.PARTITIONS_KEY_NAME, new Object[]{1})}); assignmentV100.set(ConsumerProtocol.USER_DATA_KEY_NAME, ByteBuffer.wrap(new byte[0])); assignmentV100.set("foo", "bar"); Struct headerV100 = new Struct(ConsumerProtocol.CONSUMER_PROTOCOL_HEADER_SCHEMA); headerV100.set(ConsumerProtocol.VERSION_KEY_NAME, version); ByteBuffer buffer = ByteBuffer.allocate(assignmentV100.sizeOf() + headerV100.sizeOf()); headerV100.writeTo(buffer); assignmentV100.writeTo(buffer); buffer.flip(); PartitionAssignor.Assignment assignment = ConsumerProtocol.deserializeAssignment(buffer); assertEquals(toSet(Arrays.asList(new TopicPartition("foo", 1))), toSet(assignment.partitions())); }
public static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = ASSIGNMENT_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<TopicPartition> partitions = new ArrayList<>(); for (Object structObj : struct.getArray(TOPIC_PARTITIONS_KEY_NAME)) { Struct assignment = (Struct) structObj; String topic = assignment.getString(TOPIC_KEY_NAME); for (Object partitionObj : assignment.getArray(PARTITIONS_KEY_NAME)) { Integer partition = (Integer) partitionObj; partitions.add(new TopicPartition(topic, partition)); } } return new PartitionAssignor.Assignment(partitions, userData); }
ConsumerProtocol { public static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = ASSIGNMENT_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<TopicPartition> partitions = new ArrayList<>(); for (Object structObj : struct.getArray(TOPIC_PARTITIONS_KEY_NAME)) { Struct assignment = (Struct) structObj; String topic = assignment.getString(TOPIC_KEY_NAME); for (Object partitionObj : assignment.getArray(PARTITIONS_KEY_NAME)) { Integer partition = (Integer) partitionObj; partitions.add(new TopicPartition(topic, partition)); } } return new PartitionAssignor.Assignment(partitions, userData); } }
ConsumerProtocol { public static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = ASSIGNMENT_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<TopicPartition> partitions = new ArrayList<>(); for (Object structObj : struct.getArray(TOPIC_PARTITIONS_KEY_NAME)) { Struct assignment = (Struct) structObj; String topic = assignment.getString(TOPIC_KEY_NAME); for (Object partitionObj : assignment.getArray(PARTITIONS_KEY_NAME)) { Integer partition = (Integer) partitionObj; partitions.add(new TopicPartition(topic, partition)); } } return new PartitionAssignor.Assignment(partitions, userData); }  }
ConsumerProtocol { public static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = ASSIGNMENT_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<TopicPartition> partitions = new ArrayList<>(); for (Object structObj : struct.getArray(TOPIC_PARTITIONS_KEY_NAME)) { Struct assignment = (Struct) structObj; String topic = assignment.getString(TOPIC_KEY_NAME); for (Object partitionObj : assignment.getArray(PARTITIONS_KEY_NAME)) { Integer partition = (Integer) partitionObj; partitions.add(new TopicPartition(topic, partition)); } } return new PartitionAssignor.Assignment(partitions, userData); }  static ByteBuffer serializeSubscription(PartitionAssignor.Subscription subscription); static PartitionAssignor.Subscription deserializeSubscription(ByteBuffer buffer); static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer); static ByteBuffer serializeAssignment(PartitionAssignor.Assignment assignment); }
ConsumerProtocol { public static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = ASSIGNMENT_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<TopicPartition> partitions = new ArrayList<>(); for (Object structObj : struct.getArray(TOPIC_PARTITIONS_KEY_NAME)) { Struct assignment = (Struct) structObj; String topic = assignment.getString(TOPIC_KEY_NAME); for (Object partitionObj : assignment.getArray(PARTITIONS_KEY_NAME)) { Integer partition = (Integer) partitionObj; partitions.add(new TopicPartition(topic, partition)); } } return new PartitionAssignor.Assignment(partitions, userData); }  static ByteBuffer serializeSubscription(PartitionAssignor.Subscription subscription); static PartitionAssignor.Subscription deserializeSubscription(ByteBuffer buffer); static PartitionAssignor.Assignment deserializeAssignment(ByteBuffer buffer); static ByteBuffer serializeAssignment(PartitionAssignor.Assignment assignment); static final String PROTOCOL_TYPE; static final String VERSION_KEY_NAME; static final String TOPICS_KEY_NAME; static final String TOPIC_KEY_NAME; static final String PARTITIONS_KEY_NAME; static final String TOPIC_PARTITIONS_KEY_NAME; static final String USER_DATA_KEY_NAME; static final short CONSUMER_PROTOCOL_V0; static final Schema CONSUMER_PROTOCOL_HEADER_SCHEMA; static final Schema SUBSCRIPTION_V0; static final Schema TOPIC_ASSIGNMENT_V0; static final Schema ASSIGNMENT_V0; }