mjsax commented on code in PR #23960:
URL: https://github.com/apache/kafka/pull/23960#discussion_r4418945669
##########
streams/src/main/java/org/apache/kafka/streams/processor/internals/ChangelogRecordDeserializationHelper.java:
##########
@@ -54,14 +54,26 @@ public static void applyChecksAndUpdatePosition(
if (versionHeader == null) {
return;
} else {
- switch (versionHeader.value()[0]) {
+ final byte[] version = versionHeader.value();
+ if (version == null || version.length == 0) {
+ log.warn("Changelog record has a version header with a null or
empty value. " +
+ "Skipping consistency information for this record.");
+ return;
+ }
+ switch (version[0]) {
case 0:
final Header vectorHeader =
record.headers().lastHeader(CHANGELOG_POSITION_HEADER_KEY);
if (vectorHeader == null) {
throw new StreamsException("This should not happen.
Consistency is enabled but the changelog "
+ "contains records without consistency
information.");
}
-
position.merge(PositionSerde.deserialize(ByteBuffer.wrap(vectorHeader.value())));
+ final byte[] vector = vectorHeader.value();
+ if (vector == null || vector.length == 0) {
+ log.warn("Changelog record has a position header with
a null or empty value. " +
+ "Skipping consistency information for this
record.");
+ return;
+ }
+
position.merge(PositionSerde.deserialize(ByteBuffer.wrap(vector)));
break;
default:
log.warn("Changelog records have been encoded using a
larger version than this server understands." +
Review Comment:
```suggestion
log.warn("Changelog records have been encoded using a
larger version than this server understands. " +
```
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
