KAFKA-4942 fixed the commitTimeoutMs being set before the commit actually started

[simplesteph]

this fixes KAFKA-4942

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/2365/
Test FAILed (JDK 8 and Scala 2.11).

[simplesteph]

unsure why the CI is failing, but the tests are passing on my laptop

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/2361/
Test PASSed (JDK 7 and Scala 2.10).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/2361/
Test PASSed (JDK 8 and Scala 2.12).

[gwenshap]

@ewencp or @kkonstantine: mind checking? this looks like a potentially important fix.

[ewencp]

@simplesteph Good catch, fix makes sense given the call to commitOffsets in there.

It seems that this was introduced as a regression in KAFKA-4161: KIP-89: Allow sink connectors to decouple flush and offset commit. Perhaps there is a unit test in WorkerSinkTaskTest that could catch such a regression?

[simplesteph]

@ewencp I've tried looking at writing a test for over an hour but no success as I'm not familiar with the PowerMock / EasyMock libraries. I also have no idea how to trigger two commits in a row, including a long commit that would keep the committing flag to true (it goes to false right away no matter what I do, therefore never giving me the chance to get the timestamp comparison error). Anyway, I'd love to see making it part of 0.10.2.1, but I'm afraid I won't be able to provide a test in time, if at all. I'm happy for anyone to take ownership of my 1 line of work if they're able to write a test. Sorry I couldn't be more helpful

[56quarters]

Hello,

I believe I've added a test that verifies that the change by @simplesteph is correct. I've confirmed that without his fix, this test does not pass.

Disclaimer: The test is not exactly pretty. It required spinning up an ExecutorService to replicate a long running commit.

The test is available here if someone would like to merge it into this work: smarter-travel-media@9569a04

Alternatively, I can open a new PR with this commit and @simplesteph's work if that's easier.

Please let me know if there's anything you'd like me to do or change. Thanks!

[simplesteph]

HI @56quarters . Feel free to open a PR and include my one line change :) You did most of the work, I really appreciate it

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/3151/
Test PASSed (JDK 8 and Scala 2.12).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/3150/
Test FAILed (JDK 7 and Scala 2.10).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/3156/
Test PASSed (JDK 8 and Scala 2.11).

[ewencp]

@simplesteph Could you close this one out since #2912 has been committed?

[simplesteph]

@ewencp done. Thanks for merging the other one. Badly needed fix :)

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/4966/
Test PASSed (JDK 8 and Scala 2.12).

[asfbot]

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.11/4982/
Test PASSed (JDK 7 and Scala 2.11).