Skip to content
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

Jump to bottom

fix: Make null key serialization/deserialization symmetrical #4351

Merged
merged 5 commits into from
Jan 20, 2020
Merged

fix: Make null key serialization/deserialization symmetrical #4351

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
5 commits
Select commit Hold shift + click to select a range
2c738c5
added test
Jan 16, 2020
82f2349
fix: Make sure null key is preserved during serialization/deserializa…
Jan 19, 2020
cf9a3a5
fixed merge conflict
Jan 19, 2020
a8c591b
updated test
Jan 19, 2020
aff74b0
Review updates
Jan 20, 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
16 changes: 16 additions & 0 deletions ksql-functional-tests/src/test/resources/query-validation-tests/table.json
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -74,6 +74,22 @@
{"topic": "OUTPUT", "key": "1", "value": null},
{"topic": "OUTPUT", "key": "1", "value": "2"}
]
},
{
"name": "should not blow up on null key",
"statements": [
"CREATE TABLE INPUT (ID bigint) WITH (kafka_topic='test_topic', value_format='DELIMITED');",
"CREATE TABLE OUTPUT as SELECT * FROM INPUT;"
],
"inputs": [
{"topic": "test_topic", "key": "1", "value": "1"},
{"topic": "test_topic", "key": null, "value": "2"},
{"topic": "test_topic", "key": "1", "value": "3"}
],
"outputs": [
{"topic": "OUTPUT", "key": "1", "value": "1"},
{"topic": "OUTPUT", "key": "1", "value": "3"}
]
}
]
}
5 changes: 4 additions & 1 deletion ksql-serde/src/main/java/io/confluent/ksql/serde/kafka/KafkaSerdeFactory.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -116,7 +116,7 @@ private static final class RowSerializer implements Serializer<Object> {

@Override
public byte[] serialize(final String topic, final Object struct) {
final Object value = ((Struct) struct).get(field);
final Object value = struct == null ? null : ((Struct) struct).get(field);
return delegate.serialize(topic, value);
}
}
Expand All @@ -140,6 +140,9 @@ private static final class RowDeserializer implements Deserializer<Object> {
public Struct deserialize(final String topic, final byte[] bytes) {
try {
final Object primitive = delegate.deserialize(topic, bytes);
if (primitive == null) {
return null;
}
final Struct struct = new Struct(schema);
struct.put(field, primitive);
return struct;
Expand Down
28 changes: 26 additions & 2 deletions ksql-serde/src/test/java/io/confluent/ksql/serde/kafka/KafkaSerdeFactoryTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,6 +17,7 @@

import static org.hamcrest.MatcherAssert.assertThat;
import static org.hamcrest.Matchers.is;
import static org.hamcrest.Matchers.nullValue;

import io.confluent.kafka.schemaregistry.client.SchemaRegistryClient;
import io.confluent.ksql.name.ColumnName;
Expand Down Expand Up @@ -146,8 +147,31 @@ public void shouldThrowOnValidateIfStruct() {
}

@Test
public void shouldHandleNulls() {
shouldHandle(SqlTypes.INTEGER, null);
public void shouldSerializeNullAsNull() {
// Given:
final PersistenceSchema schema = schemaWithFieldOfType(SqlTypes.INTEGER);

final Serde<Object> serde = factory.createSerde(schema, ksqlConfig, srClientFactory);

// When:
final byte[] result = serde.serializer().serialize("topic", null);

// Then:
assertThat(result, is(nullValue()));
}

@Test
public void shouldDeserializeNullAsNull() {
// Given:
final PersistenceSchema schema = schemaWithFieldOfType(SqlTypes.INTEGER);

final Serde<Object> serde = factory.createSerde(schema, ksqlConfig, srClientFactory);

// When:
final Object result = serde.deserializer().deserialize("topic", null);

// Then:
assertThat(result, is(nullValue()));
}

@Test
Expand Down