### Question:
SourceNode extends ProcessorNode<K, V> { V deserializeValue(String topic, Headers headers, byte[] data) { return valDeserializer.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();  }### Answer:
@Test public void shouldProvideTopicHeadersAndDataToValueDeserializer() { final SourceNode<String, String> sourceNode = new MockSourceNode<>(new String[]{""}, new TheExtendedDeserializer(), new TheExtendedDeserializer()); final RecordHeaders headers = new RecordHeaders(); final String deserializedValue = sourceNode.deserializeValue("topic", headers, "data".getBytes(StandardCharsets.UTF_8)); assertThat(deserializedValue, is("topic" + headers + "data")); }