Skip to content

KAFKA-10066: TestOutputTopic should pass record headers into deserializers #8759

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mjsax merged 6 commits into from
Jun 4, 2020
Merged

KAFKA-10066: TestOutputTopic should pass record headers into deserializers #8759

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
c8ec9ea
KAFKA-10066: TestOutputTopic should pass record headers into
mjsax May 29, 2020
e58a0c4
Unify test
mjsax May 29, 2020
2f2515c
foo
mjsax May 29, 2020
ef08ac8
Fix checkstyle
mjsax May 29, 2020
6e3563f
Github comment
mjsax Jun 2, 2020
32ec931
Rename test
mjsax Jun 2, 2020
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
8 changes: 4 additions & 4 deletions streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -802,8 +802,8 @@ public <K, V> ProducerRecord<K, V> readOutput(final String topic,
if (record == null) {
return null;
}
final K key = keyDeserializer.deserialize(record.topic(), record.key());
final V value = valueDeserializer.deserialize(record.topic(), record.value());
final K key = keyDeserializer.deserialize(record.topic(), record.headers(), record.key());
final V value = valueDeserializer.deserialize(record.topic(), record.headers(), record.value());
return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), key, value, record.headers());
}

Expand Down Expand Up @@ -906,8 +906,8 @@ <K, V> TestRecord<K, V> readRecord(final String topic,
if (record == null) {
throw new NoSuchElementException("Empty topic: " + topic);
}
final K key = keyDeserializer.deserialize(record.topic(), record.key());
final V value = valueDeserializer.deserialize(record.topic(), record.value());
final K key = keyDeserializer.deserialize(record.topic(), record.headers(), record.key());

@chia7712 chia7712 May 31, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nice finding!

https://github.com/apache/kafka/blob/trunk/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java#L805 has similar issue. Should we fix it as well?

@mjsax mjsax Jun 1, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

readOutput is deprecated. Thus not sure if it's worth to fix?

@chia7712 chia7712 Jun 1, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Thus not sure if it's worth to fix?

Personally, what we should add to readOutput is "deprecation" rather than "a bug". Hence, it is worthwhile to fix if the fix does not break anything.

@mjsax mjsax Jun 2, 2020

Copy link
Copy Markdown
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Not sure if I can follow? readOutput is marked as deprecated: https://github.com/apache/kafka/blob/trunk/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java#L797

I can still fix it on the side, but nobody should use it any longer and thus the gain seems minimal.

@chia7712 chia7712 Jun 2, 2020

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

but nobody should use it any longer and thus the gain seems minimal.

You are right. The benefit is too low.

final V value = valueDeserializer.deserialize(record.topic(), record.headers(), record.value());
return new TestRecord<>(key, value, record.headers(), record.timestamp());
}

Expand Down
54 changes: 54 additions & 0 deletions streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -22,10 +22,13 @@
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeader;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.ByteArraySerializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.LongDeserializer;
import org.apache.kafka.common.serialization.LongSerializer;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.SystemTime;
Expand Down Expand Up @@ -70,6 +73,7 @@
import java.util.Objects;
import java.util.Properties;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.regex.Pattern;

import static org.apache.kafka.common.utils.Utils.mkEntry;
Expand Down Expand Up @@ -711,6 +715,56 @@ public void shouldUseSourceSpecificDeserializers() {
assertThat(result2.getValue(), equalTo(source2Value));
}

@Test
public void shouldPassRecordHeadersIntoSerializersAndDeserializers() {
testDriver = new TopologyTestDriver(setupSourceSinkTopology(), config);

final AtomicBoolean passedHeadersToKeySerializer = new AtomicBoolean(false);
final AtomicBoolean passedHeadersToValueSerializer = new AtomicBoolean(false);
final AtomicBoolean passedHeadersToKeyDeserializer = new AtomicBoolean(false);
final AtomicBoolean passedHeadersToValueDeserializer = new AtomicBoolean(false);

final Serializer<byte[]> keySerializer = new ByteArraySerializer() {
@Override
public byte[] serialize(final String topic, final Headers headers, final byte[] data) {
passedHeadersToKeySerializer.set(true);
return serialize(topic, data);
}
};
final Serializer<byte[]> valueSerializer = new ByteArraySerializer() {
@Override
public byte[] serialize(final String topic, final Headers headers, final byte[] data) {
passedHeadersToValueSerializer.set(true);
return serialize(topic, data);
}
};

final Deserializer<byte[]> keyDeserializer = new ByteArrayDeserializer() {
@Override
public byte[] deserialize(final String topic, final Headers headers, final byte[] data) {
passedHeadersToKeyDeserializer.set(true);
return deserialize(topic, data);
}
};
final Deserializer<byte[]> valueDeserializer = new ByteArrayDeserializer() {
@Override
public byte[] deserialize(final String topic, final Headers headers, final byte[] data) {
passedHeadersToValueDeserializer.set(true);
return deserialize(topic, data);
}
};

final TestInputTopic<byte[], byte[]> inputTopic = testDriver.createInputTopic(SOURCE_TOPIC_1, keySerializer, valueSerializer);
final TestOutputTopic<byte[], byte[]> outputTopic = testDriver.createOutputTopic(SINK_TOPIC_1, keyDeserializer, valueDeserializer);
inputTopic.pipeInput(testRecord1);
outputTopic.readRecord();

assertThat(passedHeadersToKeySerializer.get(), equalTo(true));
assertThat(passedHeadersToValueSerializer.get(), equalTo(true));
assertThat(passedHeadersToKeyDeserializer.get(), equalTo(true));
assertThat(passedHeadersToValueDeserializer.get(), equalTo(true));
}

@Test
public void shouldUseSinkSpecificSerializers() {
final Topology topology = new Topology();
Expand Down