You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "mjsax (via GitHub)" <gi...@apache.org> on 2026/02/19 00:59:13 UTC
[PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
mjsax opened a new pull request, #21515:
URL: https://github.com/apache/kafka/pull/21515
This PR disallows to pass in Headers objects into serialize() and deserializer() to ensure we don't accidentally introduce bugs in our own code.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21515:
URL: https://github.com/apache/kafka/pull/21515#discussion_r2835471802
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ValueTimestampHeadersDeserializer.java:
##########
@@ -127,11 +132,12 @@ static <T> T value(final byte[] rawValueTimestampHeaders, final Deserializer<T>
final ByteBuffer buffer = ByteBuffer.wrap(rawValueTimestampHeaders);
final int headersSize = ByteUtils.readVarint(buffer);
- // skip headers plus timestamp
- buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] rawHeaders = readBytes(buffer, headersSize);
+ final Headers headers = HeadersDeserializer.deserialize(rawHeaders);
+ buffer.position(buffer.position() + Long.BYTES); // skip timestamp
final byte[] bytes = readBytes(buffer, buffer.remaining());
- return deserializer.deserialize("", bytes);
+ return deserializer.deserialize("", headers, bytes);
Review Comment:
One more fix... @UladzislauBlok pointed me to it! Great find.
Base on this, I am wondering about "lazy header deserialization" idea... Left a comment on https://issues.apache.org/jira/browse/KAFKA-20155
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on code in PR #21515:
URL: https://github.com/apache/kafka/pull/21515#discussion_r2825242784
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ValueTimestampHeadersDeserializer.java:
##########
@@ -69,6 +69,11 @@ public void configure(final Map<String, ?> configs, final boolean isKey) {
headersDeserializer.configure(configs, isKey);
}
+ @Override
+ public ValueTimestampHeaders<V> deserialize(final String topic, final Headers headers, final byte[] valueTimestampHeaders) {
+ throw new UnsupportedOperationException("Headers are encoded inside `valueTimestampHeaders` byte[] array and cannot be passed as parameter. Use `deserialize(String topic, byte[] valueTimestampHeaders)` instead.");
Review Comment:
Frankly, I am not 100% if this is a too restrictive check?
We could also check if `header == null || headers.toArray().size == 0` and allow the call for this case? Same for the serializer. Thoughts?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21515:
URL: https://github.com/apache/kafka/pull/21515#issuecomment-4166415637
Closing this PR. Base on more recent code changes, this does not make sense any longer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax closed pull request #21515: MINOR: Improve ValueTimestampHeader Serdes
URL: https://github.com/apache/kafka/pull/21515
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21515:
URL: https://github.com/apache/kafka/pull/21515#issuecomment-3928625387
The concern is both of the points you mentioned. -- And yes, I agree that it's not ideal... But it might be "ok-ish", given that these classes an KS internal only?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "UladzislauBlok (via GitHub)" <gi...@apache.org>.
UladzislauBlok commented on code in PR #21515:
URL: https://github.com/apache/kafka/pull/21515#discussion_r2838146099
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ValueTimestampHeadersDeserializer.java:
##########
@@ -127,11 +132,12 @@ static <T> T value(final byte[] rawValueTimestampHeaders, final Deserializer<T>
final ByteBuffer buffer = ByteBuffer.wrap(rawValueTimestampHeaders);
final int headersSize = ByteUtils.readVarint(buffer);
- // skip headers plus timestamp
- buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] rawHeaders = readBytes(buffer, headersSize);
+ final Headers headers = HeadersDeserializer.deserialize(rawHeaders);
+ buffer.position(buffer.position() + Long.BYTES); // skip timestamp
final byte[] bytes = readBytes(buffer, buffer.remaining());
- return deserializer.deserialize("", bytes);
+ return deserializer.deserialize("", headers, bytes);
Review Comment:
This method was added two weeks ago as part of #21408
I assume it was created for future use—is that correct?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on code in PR #21515:
URL: https://github.com/apache/kafka/pull/21515#discussion_r2837955511
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/ValueTimestampHeadersDeserializer.java:
##########
@@ -127,11 +132,12 @@ static <T> T value(final byte[] rawValueTimestampHeaders, final Deserializer<T>
final ByteBuffer buffer = ByteBuffer.wrap(rawValueTimestampHeaders);
final int headersSize = ByteUtils.readVarint(buffer);
- // skip headers plus timestamp
- buffer.position(buffer.position() + headersSize + Long.BYTES);
+ final byte[] rawHeaders = readBytes(buffer, headersSize);
+ final Headers headers = HeadersDeserializer.deserialize(rawHeaders);
+ buffer.position(buffer.position() + Long.BYTES); // skip timestamp
final byte[] bytes = readBytes(buffer, buffer.remaining());
- return deserializer.deserialize("", bytes);
+ return deserializer.deserialize("", headers, bytes);
Review Comment:
Was n't this the method we intended to remove as there is no call to it (no usage)
https://issues.apache.org/jira/browse/KAFKA-20193
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21515:
URL: https://github.com/apache/kafka/pull/21515#issuecomment-3936620067
@frankvicky -- You are right; technically we don't need the overload given the default impl. The idea was to make it explicit on the class itself, that both methods are not supported (also guards us for the case, that the default impl would get changed; even if I don't expect this to happen as would raise backward compatibility questions).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "aliehsaeedii (via GitHub)" <gi...@apache.org>.
aliehsaeedii commented on PR #21515:
URL: https://github.com/apache/kafka/pull/21515#issuecomment-3932758319
> The concern is both of the points you mentioned. -- And yes, I agree that it's not ideal... But it might be "ok-ish", given that these classes an KS internal only?
Ok, so let’s go ahead and be strict and throw an exception anyway, since we know this method should never be called. Thanks for being so thorough.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] MINOR: Improve ValueTimestampHeader Serdes [kafka]
Posted by "mjsax (via GitHub)" <gi...@apache.org>.
mjsax commented on PR #21515:
URL: https://github.com/apache/kafka/pull/21515#issuecomment-3942198185
Just realize that we should not merge this before https://github.com/apache/kafka/pull/21511 and rebase and update to add similar checks for `AggregationWithHeaders`(de)serializer.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org