KAFKA-12229: reset to original class loader after connector stop

[showuon]

java.lang.NullPointerException at 
org.apache.kafka.connect.mirror.MirrorSourceConnector.listTopics(MirrorSourceConnector.java:348) at 
org.apache.kafka.connect.mirror.MirrorSourceConnector.findSourceTopicPartitions(MirrorSourceConnector.java:192) at 
org.apache.kafka.connect.mirror.MirrorSourceConnectorTest.testRefreshTopicPartitionsTopicOnTargetFirst(MirrorSourceConnectorTest.java:222)

After days of investigation, I finally found the root cause of the test failure reason: class loader.

The issue is quite weird, we mocked the method, but still call the real method, and cause the NPE. Digging into the Mockito, found it's not about JUnit 5, it's because of the class loader. In Mockito, we relies on the class loader to generate the proxy instance (source) to intercept the method call, and if the class loader is not expected, we'll generate the wrong proxy instance (with wrong class path). We set the class loader during connector start to resolve conflicting dependencies (KIP-146), so we should set it back to the original class loader after connector stop in tests (EmbeddedConnectCluster is only used in tests) for the following Mockito works as expected.

So, there's an interference of integration tests with unit tests when Connect integration tests run before the MM2 unit tests, and that will cause the Mockito used in unit tests not work as expected.

Committer Checklist (excluded from commit message)

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

[showuon]

@chia7712 @ijuma @guozhangwang , please take a look at the PR. Thanks.

[showuon]

All tests passed.

[showuon]

@chia7712 @ijuma , please help review this PR. Thanks.

[kkonstantine]

Thanks for the fix @showuon
I'd like to understand a bit better what you've observed.
Are you saying there's an interference of integration tests with unit tests when Connect integration tests run before the MM2 unit tests? If that's the case, I think this connection is missing from the description. Have you noticed how often that happens (every time or occasionally)?

[showuon]

@kkonstantine , thanks for the comments. Answer your questions below:

1. Have you noticed how often that happens (every time or occasionally)?
   --> I'd say, it happens often. I just checked the kafka-trunk-jdk11, it failed the test cases from build KAFKA-2744: Commit source task offsets after task is completely stopped to ensure no additional messages are processed during the offset commit when stopping tasks for rebalancing. #423 ~ # 431 (the latest). And also happens a lot in kafka-trunk-jdk8 and kafka-trunk-jdk15. And sometimes it didn't happen is because the unit tests run faster than the integration tests.

2. Are you saying there's an interference of integration tests with unit tests when Connect integration tests run before the MM2 unit tests? If that's the case, I think this connection is missing from the description.
   --> Yes, you're right. There's an interference of integration tests with unit tests when Connect integration tests run before the MM2 unit tests. I've added in the description.

Thank you very much.

[showuon]

So, to easily reproduce it, you can try to start the EmbeddedConnectCluster before the tests in MirrorSourceConnectorTest. What I did is to copy the beforeEach startClusters() method from MirrorConnectorsIntegrationBaseTest and paste into the MirrorSourceConnectorTest. So you'll find even we mock the method, the real methods still got called.

[kkonstantine]

Thanks for the clarification @showuon
The fix makes sense. Given that we use a util method to swap the classloader in Connect in all the places that this happens, I've added an optional suggestion that is equivalent with your solution for better tracking.

Otherwise, LGTM

[showuon]

@kkonstantine , thanks for the comments and updates.
Failed tests are not related to my changes.

Test Result (2 failures / +2)

    Build / JDK 11 / org.apache.kafka.clients.consumer.KafkaConsumerTest.testCloseWithTimeUnit()
    Build / JDK 11 / org.apache.kafka.clients.consumer.internals.FetcherTest.testEarlierOffsetResetArrivesLate()

[kkonstantine]

Indeed @showuon
jdk8 and 15 are green. Thanks for the non-obvious fix.
LGTM

[showuon]

The latest kafka-trunk-jdk8, kafka-trunk-jdk11, kafka-trunk-jdk15 are not seeing these failed tests anymore. YEAH~