@Test public void testKafkaProtobufSerializer() { byte[] bytes; bytes = protobufSerializer.serialize(topic, HELLO_WORLD_MESSAGE); assertEquals(HELLO_WORLD_MESSAGE, testMessageDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, HELLO_WORLD_MESSAGE); assertEquals(HELLO_WORLD_MESSAGE, deriveTypeDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, HELLO_WORLD_MESSAGE); DynamicMessage message = (DynamicMessage) protobufDeserializer.deserialize(topic, bytes); assertEquals(HELLO_WORLD_MESSAGE.getTestString(), getField(message, "test_string")); assertEquals(HELLO_WORLD_MESSAGE.getTestInt32(), getField(message, "test_int32")); bytes = protobufSerializer.serialize(topic, message); assertEquals(HELLO_WORLD_MESSAGE, testMessageDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, message); assertEquals(HELLO_WORLD_MESSAGE, deriveTypeDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, message); message = (DynamicMessage) protobufDeserializer.deserialize(topic, bytes); assertEquals(HELLO_WORLD_MESSAGE.getTestString(), getField(message, "test_string")); assertEquals(HELLO_WORLD_MESSAGE.getTestInt32(), getField(message, "test_int32")); bytes = protobufSerializer.serialize(topic, NESTED_MESSAGE); assertEquals(NESTED_MESSAGE, nestedMessageDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, NESTED_MESSAGE); assertEquals(NESTED_MESSAGE, deriveTypeDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, NESTED_MESSAGE); message = (DynamicMessage) protobufDeserializer.deserialize(topic, bytes); assertEquals(NESTED_MESSAGE.getUserId().getKafkaUserId(), getField((DynamicMessage) getField(message, "user_id"), "kafka_user_id") ); bytes = protobufSerializer.serialize(topic, message); assertEquals(NESTED_MESSAGE, nestedMessageDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, message); assertEquals(NESTED_MESSAGE, deriveTypeDeserializer.deserialize(topic, bytes)); bytes = protobufSerializer.serialize(topic, message); message = (DynamicMessage) protobufDeserializer.deserialize(topic, bytes); assertEquals(NESTED_MESSAGE.getUserId().getKafkaUserId(), getField((DynamicMessage) getField(message, "user_id"), "kafka_user_id") ); bytes = protobufSerializer.serialize(topic, null); assertEquals(null, protobufDeserializer.deserialize(topic, bytes)); }
public KafkaProtobufSerializer() { schemaCache = new SynchronizedCache<>(new LRUCache<>(DEFAULT_CACHE_CAPACITY)); }
KafkaProtobufSerializer extends AbstractKafkaProtobufSerializer<T> implements Serializer<T> { public KafkaProtobufSerializer() { schemaCache = new SynchronizedCache<>(new LRUCache<>(DEFAULT_CACHE_CAPACITY)); } }
KafkaProtobufSerializer extends AbstractKafkaProtobufSerializer<T> implements Serializer<T> { public KafkaProtobufSerializer() { schemaCache = new SynchronizedCache<>(new LRUCache<>(DEFAULT_CACHE_CAPACITY)); } 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> { public KafkaProtobufSerializer() { schemaCache = new SynchronizedCache<>(new LRUCache<>(DEFAULT_CACHE_CAPACITY)); } 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> { public KafkaProtobufSerializer() { schemaCache = new SynchronizedCache<>(new LRUCache<>(DEFAULT_CACHE_CAPACITY)); } 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();  }