KAFKA-10531: Check for negative values to Thread.sleep call

[soondenana]

System.currentTimeMillis() is not monotonic, so using that to calculate
time to sleep can result in negative values. That will throw
IllegalArgumentException.

This change checks for that and sleeps for a second (to avoid tight
loop) if the value returned is negative.

This change need to be backported to older branches that have Connect.

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)

[ijuma]

Why don't we use a monotonic timer?

[soondenana]

Wanted to make least amount of change. I can update the code to use monotonic 'nanoTime` instead (nanoseconds in Time interface). We will also need to convert that to milli before passing to sleep (unless we want to add nano to those interfaces too, like Utils.sleep)

[soondenana]

In fact I don't like this loop altogether. Going to rewrite it so that it doesn't use these constructs.

[ijuma]

You can use hiResClockMs so that you get the value in ms.

[soondenana]

In fact we don't need value from clock to sleep, only need it to find elapsed time and timeout. I have decoupled these two, and also fixed a minor issue where it was sleeping first time even if topic was present.

Please take a look.

[stanislavkozlovski]

LGTM! Thanks for the PR!

[chia7712]

Why not using partitionsFor(String topic, Duration timeout) (https://github.com/apache/kafka/blob/trunk/clients/src/main/java/org/apache/kafka/clients/consumer/KafkaConsumer.java#L1944) to replace while loop?

[soondenana]

I think the semantics needed here is different. The timeout in partitionsFor is the max amount of time the api can block waiting for response before it fails with TimeoutException. However, the api can return within timeout with empty results as newly created topics data has not been propagated yet. We then have to retry again until partitionsFor returns the partition data (upto a max time).

[chia7712]

It seems to me the behavior of timeout is not consistent in consumer methods. The timeout used by other methods (for example: position, offsetsForTimes, beginningOffsets and endOffsets) is to await the result of specify partitions. It means consumer will send a request again if the timer is not expired and the specify partition has no metadata (i.e topics data has not been propagated yet). Maybe partitionsFor should be fixed for consistent behavior.

[soondenana]

Not sure about this, no result for "partitionInfos" is a valid result. There is no point in automatic retrying. While for other apis that retry automatically, they do it if they get an invalid result back. If null was a valid result for them, they shouldn't retry either.

[chia7712]

@soondenana Thanks for responses. This patch LGTM. What I want to discuss is unrelated to this patch.

they do it if they get an invalid result back. If null was a valid result for them, they shouldn't retry either.

If the topic is not exist (or not been propagated yet), partitionFor can return null. By contrast, beginningOffsets (and other methods) throws TimeoutException. In order to make consistent behavior, beginningOffsets (and other methods) should let the topic (or partition) be absent in the returned Map. WDYT?

[rhauch]

@soondenana if the goal is to minimize the changes, would it be sufficient to change the code to use Time.sleep(...) instead of Thread.sleep(...), and then change the SystemTime.sleep(...) implementation to return immediately if the supplied number of milliseconds is <= 0?

Are there performance implications of using System.nanoTime() instead of System.currentTimeMillis()?

[rhauch]

I have a few comments/suggestions on the current proposal, though I still wonder about the performance implication of using System.nanoTime() instead of System.currentTimeMillis().

[soondenana]

@soondenana if the goal is to minimize the changes, would it be sufficient to change the code to use Time.sleep(...) instead of Thread.sleep(...), and then change the SystemTime.sleep(...) implementation to return immediately if the supplied number of milliseconds is <= 0?

Are there performance implications of using System.nanoTime() instead of System.currentTimeMillis()?

Thanks @rhauch for taking a look. The PR started with only fixing negative sleep vlaue, but as we started looking at code more, there were multiple issue so the "minimize change" idea was dropped. Here are 3 issues with original code:

1. Negative sleep value
2. Using elapsed time since loop started to decide on next sleep time.
3. Sleeping even if the partitionsFor call would be successful for first time

Considering that I decided to rewrite the loop to fix all three issues.

Good question on performance. It seems like System.nanoTime() is slower than System.currentTimeMillis (one is read from h/w using i/o another one is reading from memory), but not sure if that has any implication here for this use case. We are waiting for topic to appear in metadata and sleeping for seconds. Few milliseconds difference should not matter.

[rhauch]

Considering that I decided to rewrite the loop to fix all three issues.

That makes sense, and I noticed the same when reviewing.

Good question on performance. It seems like System.nanoTime() is slower than System.currentTimeMillis (one is read from h/w using i/o another one is reading from memory), but not sure if that has any implication here for this use case. We are waiting for topic to appear in metadata and sleeping for seconds. Few milliseconds difference should not matter.

That's probably true. But if Thread.sleep(negativeTime) throws an IllegalArgumentException, should we also change Utils.sleep(...) (called by SystemTime.sleep(...) that is now used above) to check and return immediately for a negative number of milliseconds to sleep?

[soondenana]

should we also change Utils.sleep(...) (called by SystemTime.sleep(...) that is now used above) to check and return immediately for a negative number of milliseconds to sleep?

Yes, makes sense. Updated code to do so.

[rhauch]

LGTM. Thanks, @soondenana. And I'm fine removing the Utils.sleep(...) change, too, if that's the preferred approach.

[soondenana]

There was an error when building streams.examples:

[2020-10-05T08:40:05.722Z] [ERROR] Failed to execute goal org.apache.maven.plugins:maven-archetype-plugin:3.2.0:generate (default-cli) on project standalone-pom: A Maven project already exists in the directory /home/jenkins/workspace/Kafka_kafka-pr_PR-9347/streams/quickstart/test-streams-archetype/streams.examples -> [Help 1]

The failure is not related to this PR.