KAFKA-20127: Add new methods to StateSerdes #21411
Conversation
| * @param rawKey the key as raw bytes | ||
| * @return the key as typed object | ||
| */ | ||
| @Deprecated |
There was a problem hiding this comment.
Does that make sense to replace this deprecated method with the newly introduced one?
There was a problem hiding this comment.
I suppose that we should add @SupressWarning and replace them in follow-up PRs.
frankvicky
force-pushed
the
KAFKA-20127
branch
3 times, most recently
from
82f39db to
847d808
Compare
# Conflicts: # streams/src/main/java/org/apache/kafka/streams/state/internals/HeadersSerializer.java
frankvicky
force-pushed
the
KAFKA-20127
branch
from
847d808 to
d5fdfb2
Compare
| @@ -149,26 +151,49 @@ public String topic() { | |||
| * @param rawKey the key as raw bytes | |||
| * @return the key as typed object | |||
There was a problem hiding this comment.
Add:
@deprecated Since 4.3. Use {@link #keyFrom(byte[], Headers} instead.
Similar elsewhere.
| * @param rawValue the value as raw bytes | ||
| * @return the value as typed object | ||
| */ | ||
| public V valueFrom(final byte[] rawValue, Headers headers) { |
| * @return the value as typed object | ||
| */ | ||
| public V valueFrom(final byte[] rawValue, Headers headers) { | ||
| return valueSerde.deserializer().deserialize(topic, headers, Utils.wrapNullable(rawValue)); |
There was a problem hiding this comment.
Why do we need to add Utils.wrapNullable -- we don't use it in existing valueFrom(...) either?
| * @return the serialized key | ||
| */ | ||
| public byte[] rawKey(final K key, final Headers headers) { | ||
| try { |
There was a problem hiding this comment.
This is a lot of code duplication -- should we update existing rawKey(...) to just call this new one, passing in new RecordHeaders() object instead?
| * @return the serialized value | ||
| */ | ||
| @SuppressWarnings("rawtypes") | ||
| public byte[] rawValue(final V value, final Headers headers) { |
| } | ||
|
|
||
| @Test | ||
| public void shouldDeserializeValueWithNullHeaders() { |
| } | ||
|
|
||
| @Test | ||
| public void shouldSerializeKeyWithEmptyHeaders() { |
There was a problem hiding this comment.
This case should be covered implicitly (if we let rawKey(key) call rawKey(key, new RecordHeaders())) via the existing test for rawKey(key). So might be redundant?
| } | ||
|
|
||
| @Test | ||
| public void shouldSerializeValueWithEmptyHeaders() { |
| } | ||
|
|
||
| @Test | ||
| public void shouldThrowIfIncompatibleSerdeForKeyWithHeaders() throws ClassNotFoundException { |
| } | ||
|
|
||
| @Test | ||
| public void shouldThrowIfIncompatibleSerdeForValueWithHeaders() throws ClassNotFoundException { |
3 participants
JIRA: KAFKA-20127
This PR is part of implementation of KIP-1271, adding new methods to
enable storing headers in state store.