@Test public void testInner() { byte[] bytes; bytes = protobufSerializer.serialize(topic, INNER_MESSAGE); assertEquals(INNER_MESSAGE, innerMessageDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, INNER_MESSAGE); DynamicMessage message = (DynamicMessage) protobufDeserializer.deserialize(topic, bytes); assertEquals(INNER_MESSAGE.getId(), getField(message, "id")); }
@Override public byte[] serialize(String topic, T record) { if (record == null) { return null; } ProtobufSchema schema = schemaCache.get(record.getDescriptorForType()); if (schema == null) { schema = ProtobufSchemaUtils.getSchema(record); schemaCache.put(record.getDescriptorForType(), schema); } return serializeImpl(getSubjectName(topic, isKey, record, schema), topic, isKey, record, schema); }
KafkaProtobufSerializer extends AbstractKafkaProtobufSerializer<T> implements Serializer<T> { @Override public byte[] serialize(String topic, T record) { if (record == null) { return null; } ProtobufSchema schema = schemaCache.get(record.getDescriptorForType()); if (schema == null) { schema = ProtobufSchemaUtils.getSchema(record); schemaCache.put(record.getDescriptorForType(), schema); } return serializeImpl(getSubjectName(topic, isKey, record, schema), topic, isKey, record, schema); } }
KafkaProtobufSerializer extends AbstractKafkaProtobufSerializer<T> implements Serializer<T> { @Override public byte[] serialize(String topic, T record) { if (record == null) { return null; } ProtobufSchema schema = schemaCache.get(record.getDescriptorForType()); if (schema == null) { schema = ProtobufSchemaUtils.getSchema(record); schemaCache.put(record.getDescriptorForType(), schema); } return serializeImpl(getSubjectName(topic, isKey, record, schema), topic, isKey, record, schema); } KafkaProtobufSerializer();  KafkaProtobufSerializer(SchemaRegistryClient client);  KafkaProtobufSerializer(SchemaRegistryClient client, Map<String, ?> props);  KafkaProtobufSerializer(SchemaRegistryClient client, Map<String, ?> props,
                                 int cacheCapacity); }
KafkaProtobufSerializer extends AbstractKafkaProtobufSerializer<T> implements Serializer<T> { @Override public byte[] serialize(String topic, T record) { if (record == null) { return null; } ProtobufSchema schema = schemaCache.get(record.getDescriptorForType()); if (schema == null) { schema = ProtobufSchemaUtils.getSchema(record); schemaCache.put(record.getDescriptorForType(), schema); } return serializeImpl(getSubjectName(topic, isKey, record, schema), topic, isKey, record, schema); } KafkaProtobufSerializer();  KafkaProtobufSerializer(SchemaRegistryClient client);  KafkaProtobufSerializer(SchemaRegistryClient client, Map<String, ?> props);  KafkaProtobufSerializer(SchemaRegistryClient client, Map<String, ?> props,
                                 int cacheCapacity); @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, T record); @Override void close(); }
KafkaProtobufSerializer extends AbstractKafkaProtobufSerializer<T> implements Serializer<T> { @Override public byte[] serialize(String topic, T record) { if (record == null) { return null; } ProtobufSchema schema = schemaCache.get(record.getDescriptorForType()); if (schema == null) { schema = ProtobufSchemaUtils.getSchema(record); schemaCache.put(record.getDescriptorForType(), schema); } return serializeImpl(getSubjectName(topic, isKey, record, schema), topic, isKey, record, schema); } KafkaProtobufSerializer();  KafkaProtobufSerializer(SchemaRegistryClient client);  KafkaProtobufSerializer(SchemaRegistryClient client, Map<String, ?> props);  KafkaProtobufSerializer(SchemaRegistryClient client, Map<String, ?> props,
                                 int cacheCapacity); @Override void configure(Map<String, ?> configs, boolean isKey); @Override byte[] serialize(String topic, T record); @Override void close();  }