KAFKA-12380 Executor in Connect's Worker is not shut down when the worker is

[karuturi]

added shutdown for executors and tests for the same

Committer Checklist (excluded from commit message)

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

[karuturi]

I could not find the exact failed testcases. I can see assertion errors in StreamsUncaughtExceptionHandlerIntegrationTest. If I run it locally, it's taking forever.

Can someone help me find the failed tests?
The change I added is very isolated. Should not cause any failures ideally.

[karuturi]

I downloaded the console log and a grep showed below tests

cat consoleText.txt| grep FAILED
[2022-03-28T14:02:18.545Z] SocketServerTest > testIdleConnection() FAILED
[2022-03-28T14:44:11.437Z] TransactionsBounceTest > testWithGroupId() FAILED
[2022-03-28T14:59:52.526Z] DynamicConnectionQuotaTest > testDynamicListenerConnectionCreationRateQuota() FAILED
[2022-03-28T15:04:10.646Z] KRaftClusterTest > testLegacyAlterConfigs() FAILED

all of them are successful if ran individually

[showuon]

@karuturi , thanks for the PR. But I found there was a long pending PR that is addressing the same issue (#10337). Could we wait for a few days to see if the author wants to continue the old PR or not, before we start review this PR? (I've left a comment in that PR).

Thanks.

[karuturi]

@showuon Sure, can wait.
Hope we will merge one of them soon :)

[karuturi]

Hi @showuon
As there is no response on the other PR, can we take this forward?

[karuturi]

@showuon Can we merge this please?

[showuon]

LGTM! Left one comment. And thanks for the tests.

[showuon]

Thanks for the update. Looks better. One minor comment left.

[showuon]

LGTM! Thanks for the patch. Let's wait for the jenkins build results.

[karuturi]

Jenkins has failures unrelated to this PR

[showuon]

Failed tests are unrelated:

Build / JDK 8 and Scala 2.12 / kafka.network.ConnectionQuotasTest.testListenerConnectionRateLimitWhenActualRateAboveLimit()

[showuon]

@karuturi , thanks for the contribution!

[karuturi]

Thanks for your time @showuon