KAFKA-8412: Still a nullpointer exception thrown on shutdown whi…

[cpettitt-confluent]

…le flushing before closing producers

Prior to this change an NPE is raised when calling AssignedTasks.close
under the following conditions:

1. EOS is enabled
2. The task was in a suspended state

The cause for the NPE is that when a clean close is requested for a
StreamTask the StreamTask tries to commit. However, in the suspended
state there is no producer so ultimately an NPE is thrown for the
contained RecordCollector in flush.

It is my opinion that in the long term, this (and probably other
surprising state interactions) could be cleaned up by consolidating
state into one place instead of spreading it across AssignedTasks,
StreamTask, and AbstractTask. However, that is a much larger, more risky
change, and this issue is currently considered minor.

The fix put forth in this commit is to have AssignedTasks call
closeSuspended when it knows the underlying StreamTask is suspended.

Currently the only externally visible way to detect this problem in test
seems to be via logging. This is because the NPE is logged but then
suppressed under the following sequence:

RecordCollectorImpl.flush:266
- throws NPE (producer is null)

StreamTask.suspend:578
- goes through the finally block and then reraises the NPE

StreamTask.close:706
- catches the NPE, calls closeSuspended with the NPE

StreamTask.closeSuspended:676
- rethrows the NPE after some cleanup

AssignedTasks.close:341
- catches and logs the exception
- tries a "dirty" close (clean = true) which succeeds
- firstException is NOT set because the test !closeUnclean(task)
does not hold.

It seems this is not the intended behavior? If so, I will happily
correct that and stop using logging as a way to detect failure.

Otherwise this commit does not currently pass checkstyle because I'm
using blacklisted imports: LogCaptureAppender and its various
dependencies from log4j. I would appreciate guidance as to whether we
should whitelist these or use another technique for detection.

Note also that this test is quite involved. I could have just tested
that AssignedTasks calls closeSuspended when appropriate, but that is
testing, IMO, a detail of the implementation and doesn't actually verify
we reproduced the original problem as it was described. I feel much more
confident that we are reproducing the behavior - and we can test exactly
the conditions that lead to it - when testing across AssignedTasks and
StreamTask. I believe this is an additional support for the argument of
eventually consolidating the state split across classes.

More detailed description of your change,
if necessary. The PR title and PR message become
the squashed commit message, so use a separate
comment to ping reviewers.

Summary of testing strategy (including rationale)
for the feature or bug fix. Unit and/or integration
tests are expected for any behaviour change and
system tests should be considered for larger changes.

Committer Checklist (excluded from commit message)

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

[cpettitt-confluent]

@mjsax @guozhangwang Please see patch notes, particularly about how the NPE is suppressed and thus difficult to test for. I offered a few ideas, but would appreciate your guidance.

[cpettitt-confluent]

bump

[mjsax]

Why not leaf the code "as-is" and change this line to:

if (suspended.values().contains(task)) {
    task.closeSuspended(clean, false, firstException);
} else {
    task.close(clean, false);
}

[cpettitt-confluent]

Absolutely! Will do.

[mjsax]

I think the overall approach is correct. Side notice: should we revert #5993 that seems not to be necessary after this change any longer?

[cpettitt-confluent]

@mjsax

Will give the revert for #5993 a shot and verify.

BTW, I cannot commit this as is because I'm using blacklisted classes (log4j, LogCaptureAppender) in the test code. Would you be ok with me whitelisting them?

[mjsax]

I guess you can use org.apache.kafka.streams.processor.internals.testutil.LogCaptureAppender instead? Or should we remove this class and switch to kafka.utils.LogCaptureAppender? \cc @vvcephei who introduced the streams LogCaptureAppender (why did you not use the Kafka utils one?)

[cpettitt-confluent]

I inlined closeTask per your suggestion - it's definitely an improvement.

For now I whitelisted kafka.utils.LogCaptureAppender. I actually like @vvcephei 's version better - it's more idiomatic (pure Java) and hides the log4j dependency. I think the only gap is the ability to restore the log level back to its original state. I would be happy to switch over and add this functionality if preferred by whoever reviews this commit.

[guozhangwang]

LGTM. Triggering one system test before merging.

[guozhangwang]

System test passed, merging to trunk.

[guozhangwang]

Cherry-picked to 2.3, 2.2, and 2.1 as well.

Thanks @cpettitt-confluent for your contribution.