KAFKA-14491: [8/N] Add serdes for ValueAndTimestamp with null value #13249

vcrfxia · 2023-02-15T02:33:53Z

As part of introducing versioned key-value stores in KIP-889, we'd like a way to represent a versioned key-value store (VersionedKeyValueStore<Bytes, byte[]>) as a regular key-value store (KeyValueStore<Bytes, byte[]>) in order to be compatible with existing DSL methods for passing key-value stores, e.g., StreamsBuilder#table() and KTable methods, which are explicitly typed to accept Materialized<K, V, KeyValueStore<Bytes, byte[]>. This way, we do not need to introduce new versions of all relevant StreamsBuilder and KTable methods to relax the Materialized type to accept versioned stores.

As part of representing a versioned key-value store as a regular key-value bytes store, we need a way to serialize a value and timestamp as a single byte array, where the value may be null (in order to represent putting a tombstone with timestamp into the versioned store). This PR introduces the serdes for doing so. See #13250 for its usage.

Committer Checklist (excluded from commit message)

Verify design and implementation
Verify test coverage and CI build status
Verify documentation (including upgrade notes)

mjsax · 2023-02-15T18:15:36Z

...s/src/main/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerde.java

+            @Override
+            public byte[] serialize(final String topic, final Boolean data) {
+                if (data == null) {
+                    // actually want to return null here but spotbugs won't allow deserialization so


Not sure if I can follow? We should return null and just make sure spotbug does not mess with us. What does it complain about?

Resolved below.

mjsax · 2023-02-15T18:30:09Z

...s/src/main/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerde.java

+            @Override
+            public Boolean deserialize(final String topic, final byte[] data) {
+                if (data == null) {
+                    // actually want to return null here but spotbugs won't allow it (NP_BOOLEAN_RETURN_NULL)


Ah. Here is the error. We should just add an exception for this error (cf gradle/spotbugs-exclude.xml)

OK, added an exception for this particular class.

mjsax · 2023-02-15T18:37:54Z

.../main/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerializer.java

+
+        final byte[] nonNullRawValue = rawValue == null ? new byte[0] : rawValue;
+        return ByteBuffer
+            .allocate(rawTimestamp.length + rawIsTombstone.length + nonNullRawValue.length)


Both rawTimestamp.length and rawIsTombstone.length are constants, right? Should we do allocate(8L + 1L + nonNullRawValueLength)

Yes that's correct. I extracted the expected lengths into static variables and added the relevant checks in order to be defensive against serialization surprises.

mjsax · 2023-02-15T18:41:17Z

...ain/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampDeserializer.java

+            return ValueAndTimestamp.makeAllowNullable(null, timestamp);
+        } else {
+            final V value = valueDeserializer.deserialize(topic, rawValue(rawValueAndTimestamp));
+            return ValueAndTimestamp.makeAllowNullable(value, timestamp);


Should we call make() here? value should never be null for this case?

You're correct that value should never be null here, assuming valid deserializer implementations which never deserialize non-null into null.

Regarding updating this call to make(), make() does not throw if value == null, instead it returns null. In the event of a buggy deserializer implementation which returns null when it shouldn't, calling make() here and returning null will cause cascading failures elsewhere -- for example, the changelogging layer uses this deserializer to obtain the value to write to the changelog (see #13251) and will throw an exception if it gets a null ValueAndTimestamp from the deserializer.

I suppose I can update this to make() and throw an explicit exception if value == null. I'll do that.

mjsax · 2023-02-15T18:46:48Z

...c/test/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerdeTest.java

+    }
+
+    @Test
+    public void shouldSerializeNonNullWithEmptyBytes() {


The test does not really verify if we get "empty bytes"? I am also not sure if "" is actually empty, as it would encode "string length as zero" ?

I believe you want to test the boolean isTombstone flag here? Mabye we need a custom Serde for the value to force an empty byte array as serialization output?

StringSerializer serializes an empty string as byte[0], which is exactly what this test case is for (isTombstone will be set to false, which is how byte[0] is distinguished from a null value).

Did I misunderstand your question?

Yes, that was the question. I was just not sure if it would do this. SG.

vcrfxia

Thanks as always for your review, @mjsax ! Pushed a new commit just now with the requested changes.

vcrfxia · 2023-02-15T22:36:33Z

...ain/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampDeserializer.java

+            return ValueAndTimestamp.makeAllowNullable(null, timestamp);
+        } else {
+            final V value = valueDeserializer.deserialize(topic, rawValue(rawValueAndTimestamp));
+            return ValueAndTimestamp.makeAllowNullable(value, timestamp);


You're correct that value should never be null here, assuming valid deserializer implementations which never deserialize non-null into null.

Regarding updating this call to make(), make() does not throw if value == null, instead it returns null. In the event of a buggy deserializer implementation which returns null when it shouldn't, calling make() here and returning null will cause cascading failures elsewhere -- for example, the changelogging layer uses this deserializer to obtain the value to write to the changelog (see #13251) and will throw an exception if it gets a null ValueAndTimestamp from the deserializer.

I suppose I can update this to make() and throw an explicit exception if value == null. I'll do that.

vcrfxia · 2023-02-15T22:47:55Z

...s/src/main/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerde.java

+            @Override
+            public Boolean deserialize(final String topic, final byte[] data) {
+                if (data == null) {
+                    // actually want to return null here but spotbugs won't allow it (NP_BOOLEAN_RETURN_NULL)


OK, added an exception for this particular class.

vcrfxia · 2023-02-15T22:48:05Z

...s/src/main/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerde.java

+            @Override
+            public byte[] serialize(final String topic, final Boolean data) {
+                if (data == null) {
+                    // actually want to return null here but spotbugs won't allow deserialization so


Resolved below.

vcrfxia · 2023-02-15T23:02:30Z

.../main/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerializer.java

+
+        final byte[] nonNullRawValue = rawValue == null ? new byte[0] : rawValue;
+        return ByteBuffer
+            .allocate(rawTimestamp.length + rawIsTombstone.length + nonNullRawValue.length)


Yes that's correct. I extracted the expected lengths into static variables and added the relevant checks in order to be defensive against serialization surprises.

vcrfxia · 2023-02-15T23:06:12Z

...c/test/java/org/apache/kafka/streams/state/internals/NullableValueAndTimestampSerdeTest.java

+    }
+
+    @Test
+    public void shouldSerializeNonNullWithEmptyBytes() {


StringSerializer serializes an empty string as byte[0], which is exactly what this test case is for (isTombstone will be set to false, which is how byte[0] is distinguished from a null value).

Did I misunderstand your question?

add serdes for ValueAndTimestamp with null value

e6f2969

This was referenced Feb 15, 2023

KAFKA-14491: [9/N] Add versioned bytes store and supplier #13250

Merged

KAFKA-14491: [10/N] Add changelogging wrapper for versioned stores #13251

Merged

KAFKA-14491: [11/N] Add metered wrapper for versioned stores #13252

Merged

mjsax added streams kip Requires or implements a KIP labels Feb 15, 2023

mjsax reviewed Feb 15, 2023

View reviewed changes

review feedback

b093726

vcrfxia commented Feb 15, 2023

View reviewed changes

mjsax merged commit dcaf95a into apache:trunk Feb 16, 2023

vcrfxia deleted the kip-889-nullable-serde branch February 16, 2023 16:12

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

KAFKA-14491: [8/N] Add serdes for ValueAndTimestamp with null value #13249

KAFKA-14491: [8/N] Add serdes for ValueAndTimestamp with null value #13249

vcrfxia commented Feb 15, 2023 •

edited

Loading

mjsax Feb 15, 2023

vcrfxia Feb 15, 2023

mjsax Feb 15, 2023

vcrfxia Feb 15, 2023

mjsax Feb 15, 2023

vcrfxia Feb 15, 2023

mjsax Feb 15, 2023

vcrfxia Feb 15, 2023

mjsax Feb 15, 2023

vcrfxia Feb 15, 2023

mjsax Feb 16, 2023

vcrfxia left a comment

vcrfxia Feb 15, 2023

vcrfxia Feb 15, 2023

vcrfxia Feb 15, 2023

vcrfxia Feb 15, 2023

vcrfxia Feb 15, 2023

KAFKA-14491: [8/N] Add serdes for ValueAndTimestamp with null value #13249

KAFKA-14491: [8/N] Add serdes for ValueAndTimestamp with null value #13249

Conversation

vcrfxia commented Feb 15, 2023 • edited Loading

Committer Checklist (excluded from commit message)

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

vcrfxia commented Feb 15, 2023 •

edited

Loading