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

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 4 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"}
]
}
]
}
4 changes: 2 additions & 2 deletions 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 @@ -142,7 +142,7 @@ public Struct deserialize(final String topic, final byte[] bytes) {
final Object primitive = delegate.deserialize(topic, bytes);
final Struct struct = new Struct(schema);
struct.put(field, primitive);
return struct;
return primitive == null ? null : struct;
Copy link
Contributor

Choose a reason for hiding this comment

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

nit: if primitive is null, then could return after line 142.

} catch (final Exception e) {
throw new SerializationException(
"Error deserializing DELIMITED message from topic: " + topic, e);
Expand Down
8 changes: 7 additions & 1 deletion 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 @@ -148,6 +149,7 @@ public void shouldThrowOnValidateIfStruct() {
@Test
public void shouldHandleNulls() {
shouldHandle(SqlTypes.INTEGER, null);
shouldHandle(SqlTypes.STRING, null);
}

@Test
Expand Down Expand Up @@ -192,7 +194,11 @@ private void shouldHandle(final SqlType fieldSchema, final Object value) {
final Object result = serde.deserializer().deserialize("topic", bytes);

// Then:
assertThat(result, is(struct));
if (value == null) {
assertThat(result, is(nullValue()));
} else {
assertThat(result, is(struct));
}
Copy link
Contributor

Choose a reason for hiding this comment

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

if its null, then assert its null... weird asserting! ;)

Also, this means all of the tests will pass if deserialize always returns null.

I think it would be more explicit if we revert this change and instead have explicit tests that test we:

a) serialize null as null , and
b) deserialize null as null.

Just played around with this, so may as well share the code.

i.e. we revert this to:

  assertThat(result, is(struct));

And replace shouldHandleNulls with:

@Test
  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()));
  }

}

private static PersistenceSchema schemaWithFieldOfType(final SqlType fieldSchema) {
Expand Down