@Test public void deserializeNewSubscriptionVersion() { short version = 100; Schema subscriptionSchemaV100 = new Schema( new Field(ConsumerProtocol.TOPICS_KEY_NAME, new ArrayOf(Type.STRING)), new Field(ConsumerProtocol.USER_DATA_KEY_NAME, Type.BYTES), new Field("foo", Type.STRING)); Struct subscriptionV100 = new Struct(subscriptionSchemaV100); subscriptionV100.set(ConsumerProtocol.TOPICS_KEY_NAME, new Object[]{"topic"}); subscriptionV100.set(ConsumerProtocol.USER_DATA_KEY_NAME, ByteBuffer.wrap(new byte[0])); subscriptionV100.set("foo", "bar"); Struct headerV100 = new Struct(ConsumerProtocol.CONSUMER_PROTOCOL_HEADER_SCHEMA); headerV100.set(ConsumerProtocol.VERSION_KEY_NAME, version); ByteBuffer buffer = ByteBuffer.allocate(subscriptionV100.sizeOf() + headerV100.sizeOf()); headerV100.writeTo(buffer); subscriptionV100.writeTo(buffer); buffer.flip(); Subscription subscription = ConsumerProtocol.deserializeSubscription(buffer); assertEquals(Arrays.asList("topic"), subscription.topics()); }
public static PartitionAssignor.Subscription deserializeSubscription(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = SUBSCRIPTION_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<String> topics = new ArrayList<>(); for (Object topicObj : struct.getArray(TOPICS_KEY_NAME)) topics.add((String) topicObj); return new PartitionAssignor.Subscription(topics, userData); }
ConsumerProtocol { public static PartitionAssignor.Subscription deserializeSubscription(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = SUBSCRIPTION_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<String> topics = new ArrayList<>(); for (Object topicObj : struct.getArray(TOPICS_KEY_NAME)) topics.add((String) topicObj); return new PartitionAssignor.Subscription(topics, userData); } }
ConsumerProtocol { public static PartitionAssignor.Subscription deserializeSubscription(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = SUBSCRIPTION_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<String> topics = new ArrayList<>(); for (Object topicObj : struct.getArray(TOPICS_KEY_NAME)) topics.add((String) topicObj); return new PartitionAssignor.Subscription(topics, userData); }  }
ConsumerProtocol { public static PartitionAssignor.Subscription deserializeSubscription(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = SUBSCRIPTION_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<String> topics = new ArrayList<>(); for (Object topicObj : struct.getArray(TOPICS_KEY_NAME)) topics.add((String) topicObj); return new PartitionAssignor.Subscription(topics, 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.Subscription deserializeSubscription(ByteBuffer buffer) { Struct header = CONSUMER_PROTOCOL_HEADER_SCHEMA.read(buffer); Short version = header.getShort(VERSION_KEY_NAME); checkVersionCompatibility(version); Struct struct = SUBSCRIPTION_V0.read(buffer); ByteBuffer userData = struct.getBytes(USER_DATA_KEY_NAME); List<String> topics = new ArrayList<>(); for (Object topicObj : struct.getArray(TOPICS_KEY_NAME)) topics.add((String) topicObj); return new PartitionAssignor.Subscription(topics, 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; }