@Test public void shouldProvideTopicHeadersAndDataToKeyDeserializer() { final SourceNode<String, String> sourceNode = new MockSourceNode<>(new String[]{""}, new TheExtendedDeserializer(), new TheExtendedDeserializer()); final RecordHeaders headers = new RecordHeaders(); final String deserializeKey = sourceNode.deserializeKey("topic", headers, "data".getBytes(StandardCharsets.UTF_8)); assertThat(deserializeKey, is("topic" + headers + "data")); }
K deserializeKey(String topic, Headers headers, byte[] data) { return keyDeserializer.deserialize(topic, headers, data); }
SourceNode extends ProcessorNode<K, V> { K deserializeKey(String topic, Headers headers, byte[] data) { return keyDeserializer.deserialize(topic, headers, data); } }
SourceNode extends ProcessorNode<K, V> { K deserializeKey(String topic, Headers headers, byte[] data) { return keyDeserializer.deserialize(topic, headers, data); } SourceNode(String name, List<String> topics, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer);  SourceNode(String name, List<String> topics, Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer); }
SourceNode extends ProcessorNode<K, V> { K deserializeKey(String topic, Headers headers, byte[] data) { return keyDeserializer.deserialize(topic, headers, data); } SourceNode(String name, List<String> topics, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer);  SourceNode(String name, List<String> topics, Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer); @SuppressWarnings("unchecked") @Override void init(ProcessorContext context); @Override void process(final K key, final V value); @Override String toString(); String toString(String indent); TimestampExtractor getTimestampExtractor(); }
SourceNode extends ProcessorNode<K, V> { K deserializeKey(String topic, Headers headers, byte[] data) { return keyDeserializer.deserialize(topic, headers, data); } SourceNode(String name, List<String> topics, TimestampExtractor timestampExtractor, Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer);  SourceNode(String name, List<String> topics, Deserializer<K> keyDeserializer, Deserializer<V> valDeserializer); @SuppressWarnings("unchecked") @Override void init(ProcessorContext context); @Override void process(final K key, final V value); @Override String toString(); String toString(String indent); TimestampExtractor getTimestampExtractor();  }