ableegoldman commented on a change in pull request #9186:
URL: https://github.com/apache/kafka/pull/9186#discussion_r475800482
##########
File path:
streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamKTableJoinProcessor.java
##########
@@ -58,29 +60,46 @@ public void init(final ProcessorContext context) {
@Override
public void process(final K1 key, final V1 value) {
- // we do join iff keys are equal, thus, if key is null we cannot join
and just ignore the record
- // If {@code keyMapper} returns {@code null} it implies there is no
match,
- // so ignore unless it is a left join
+ // we do join iff the joining keys are equal, thus, if the mappedKey
is null we cannot join
+ // and just ignore the record.
//
// we also ignore the record if value is null, because in a key-value
data model a null-value indicates
// an empty message (ie, there is nothing to be joined) -- this
contrast SQL NULL semantics
// furthermore, on left/outer joins 'null' in ValueJoiner#apply()
indicates a missing record --
// thus, to be consistent and to avoid ambiguous null semantics, null
values are ignored
- if (key == null || value == null) {
+ final Optional<K2> maybeMappedKey = maybeExtractMappedKey(key, value);
+ if (!maybeMappedKey.isPresent()) {
LOG.warn(
"Skipping record due to null key or value. key=[{}] value=[{}]
topic=[{}] partition=[{}] offset=[{}]",
key, value, context().topic(), context().partition(),
context().offset()
);
droppedRecordsSensor.record();
} else {
- final K2 mappedKey = keyMapper.apply(key, value);
- final V2 value2 = mappedKey == null ? null :
getValueOrNull(valueGetter.get(mappedKey));
+ final K2 mappedKey = maybeMappedKey.get();
+ final V2 value2 = getValueOrNull(valueGetter.get(mappedKey));
if (leftJoin || value2 != null) {
context().forward(key, joiner.apply(value, value2));
}
}
}
+ private Optional<K2> maybeExtractMappedKey(final K1 key, final V1 value) {
+ if (value == null) {
+ return Optional.empty();
+ }
+
+ // we allow the case where the key is null but mappedKey is not null
and thus
+ // we need to guard against nullPointerExceptions. This may happen for
GlobalKTables.
+ // For KTables, the keyMapper simply returns the key, so this will
never happen
+ Optional<K2> maybeMappedKey;
+ try {
+ maybeMappedKey = Optional.ofNullable(keyMapper.apply(key, value));
+ } catch (final NullPointerException e) {
Review comment:
Yeah the compatibility argument is reasonable, but you could say that
users should have been handling the null case all along and it was just due to
a bug in Streams that we never actually passed in a null key. If people
_aren't_ handling null, we should try and alert them quickly (and nothing
catches people's attention faster than an NPE)
----------------------------------------------------------------
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.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
