MINOR: Prohibit setting StreamsConfig commit.interval.ms to a negative value

[occho]

Prohibit setting StreamsConfig commit.interval.ms to a negative value to avoid possible ambiguity of what it indicates.

So far, setting the property to a negative value can be used to turn off periodic offset-commit by StreamThread. With this change, that will no longer work.
Instead of using a negative value, LONG.MAX_VALUE can be used to achieve the almost same thing.

Committer Checklist (excluded from commit message)

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

[occho]

We might be able to remove the check interval >= 0, but I'm not sure whether we can assume their value always come from StreamsConfig. So, I did not update the code.

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamThread.java

Line 1023 in 9951f8f

if (commitTimeMs >= 0 && now - lastCommitMs > commitTimeMs) {

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

Line 244 in 9951f8f

if (flushInterval >= 0 && now >= lastFlush + flushInterval) {

[occho]

@guozhangwang Could you review this?

[guozhangwang]

LGTM. This is from a discussion on the accepted range of this config, cc @bbejeck @vvcephei @mjsax for another review.

Although it changes the public config's range, I think this is really a "bug" that we should fix rather than changing the semantics, so we can skip the KIP process.

[mjsax]

It's a judgement call. I am fine with the change. We recently updated KafkaStreams#close() similarly.

@occho I think it's save the remove the check >= 0 in both classes. Can you update the code accordingly?

[occho]

@occho I think it's save the remove the check >= 0 in both classes. Can you update the code accordingly?

Sure. Thank you for your advice.
Updated.

[bbejeck]

Overall looks good to me. I think we should have some unit tests demonstrating the fix i.e an Exception results when trying to set commit.interval.ms to a negative value

[occho]

Added a test.

[occho]

Ah, wait, I need more fix.

[occho]

Updated and rebased onto the latest trunk.

[bbejeck]

Thanks @occho just one very minor nit, otherwise LGTM

[bbejeck]

nit: maybe keep this test but pass in 0 instead?

[occho]

From my understanding, the test name is wrong and should be like shouldNotFlushWhenFlushIntervalIsNegative.
The code that shouldNotFlushWhenFlushIntervalIsZero tests is about this line:

kafka/streams/src/main/java/org/apache/kafka/streams/processor/internals/GlobalStreamThread.java

Line 244 in 74f686d

if (flushInterval >= 0 && now >= lastFlush + flushInterval) {

With the code on trunk, when flushInterval is set to negative, stateMaintainer.flushState() is not called.
But, since this spec is removed, I think it is okay to remove the test.

WDYT?

[bbejeck]

Ack, thanks for the explanation. Sounds good to me.

[bbejeck]

retest this please

[mjsax]

LGTM.

Thanks for the PR @occho!

[occho]

Thank you for reviewing. :)