KAFKA-15845: Detect leaked Kafka clients and servers with LeakTestingExtension

[gharris1727]

There are currently many tests which instantiate sockets and channels, but don't close them. The vast majority of these sockets and channels are created via Selector and SocketServer, and are associated with clients and servers which are not closed properly. Many of these leaks are silent, and have gone unnoticed for years.

To allow us to detect these leaks and prevent future ones, we should have a Junit5 Extension which can detect when tests leak clients, and fail those tests with a diagnostic message.

Implementation note: I tried to keep the main changes as small as possible, making use of the javax.net Factory classes to create a layer of indirection in the NetworkContext. This is a much less invasive change than dependency-injecting the factories and possibly having to add more constructors to the Kafka clients.

While trying this out, I applied it automatically to all test suites with Jupiter's automatic registration mechanism. I added opt-out IgnoreAll and IgnoreEach annotations. I found that many test suites create a single Kafka cluster for the whole test, and this would cause the Each extension to generate false positives. This lead me to believe that Each should be opt-in instead, and a developer can add it temporarily with @ExtendWith(LeakTestingExtension.Each.class).

However, using automatic registration also means that consumers of the clients test-jar may unintentionally turn on this leak testing. We could avoid this by moving the extension to a different test-utils jar that is only for internal-use, or by disabling automatic registration and making this extension opt-in only.

This PR will cause many test failures for tests that are not compliant, and should not be merged as-is. For the set of leaks I found with this testing methodology, see:

• KAFKA-15804: Close SocketServer channels when calling shutdown before enableRequestProcessing #14729
• KAFKA-15816: Fix leaked sockets in clients tests #14750
• KAFKA-15819: Fix leaked KafkaRaftManager in ZK mode during migration #14751
• KAFKA-15816: Fix leaked sockets in core tests #14754
• KAFKA-15816: Fix leaked sockets in mirror tests #14761
• KAFKA-15826: Close consumer when sink task is cancelled #14762
• KAFKA-15827: Prevent KafkaBasedLog subclasses from leaking passed-in clients #14763
• KAFKA-15816: Fix leaked sockets in runtime tests #14764
• KAFKA-15816: Fix leaked sockets in streams tests #14769
• KAFKA-15816: Fix leaked sockets in trogdor tests #14771
• KAFKA-15834: Remove NamedTopologyIntegrationTest which leaks clients #15185
• KAFKA-15834: Remove more leaky NamedTopologyIntegrationTest tests #15243
• KAFKA-16702: Fix producer leaks in KafkaLog4jAppenderTest #15922
• KAFKA-16703: Close serverChannel in SocketServer if unable to bind to a port #15923
• KAFKA-16765: NioEchoServer leaks accepted SocketChannel instances due to race condition #16576
• KAFKA-17204: KafkaStreamsCloseOptionsIntegrationTest.before leaks AdminClient #16692
• KAFKA-17202: EosIntegrationTest.verifyChangelogMaxRecordOffsetMatchesCheckpointedOffset leaks consumers #16702
• KAFKA-17201: SelectorTest.testInboundConnectionsCountInConnectionCreationMetric leaks sockets and threads #16723
• KAFKA-17203: StreamThread leaking producer instances #16730
• https://issues.apache.org/jira/browse/KAFKA-16768
• https://issues.apache.org/jira/browse/KAFKA-17207

If you have a leaked resource, it looks something like this (from KafkaProducerTest):

This test contains a resource leak. Close the resources instantiated in the test, or add the @LeakTestingExtension.Ignore annotation to the method and/or class.
java.lang.AssertionError: This test contains a resource leak. Close the resources instantiated in the test, or add the @LeakTestingExtension.Ignore annotation to the method and/or class.
	at org.apache.kafka.common.network.LeakTestingExtension.after(LeakTestingExtension.java:82)
	at org.apache.kafka.common.network.LeakTestingExtension.access$100(LeakTestingExtension.java:31)
	at org.apache.kafka.common.network.LeakTestingExtension$Each.afterEach(LeakTestingExtension.java:125)
	at org.junit.jupiter.engine.descriptor.TestMethodTestDescriptor.lambda$invokeAfterEachCallbacks$12(TestMethodTestDescriptor.java:261)
<snipped>
	at worker.org.gradle.process.internal.worker.GradleWorkerMain.run(GradleWorkerMain.java:69)
	at worker.org.gradle.process.internal.worker.GradleWorkerMain.main(GradleWorkerMain.java:74)
Caused by: java.lang.AssertionError: Leak check failed
	at org.apache.kafka.common.utils.LeakTester.lambda$combine$0(LeakTester.java:87)
	at org.apache.kafka.common.network.LeakTestingExtension.after(LeakTestingExtension.java:80)
	... 70 more
	Suppressed: java.lang.AssertionError: AbstractSelector instances left open
		at org.apache.kafka.common.utils.PredicateLeakTester.lambda$start$0(PredicateLeakTester.java:90)
		at org.apache.kafka.common.utils.LeakTester.lambda$combine$0(LeakTester.java:84)
		... 71 more
		Suppressed: java.lang.Exception: Opened sun.nio.ch.KQueueSelectorImpl
			at org.apache.kafka.common.utils.PredicateLeakTester.open(PredicateLeakTester.java:59)
			at org.apache.kafka.common.network.NetworkContextLeakTester$RecordingSelectorProvider.openSelector(NetworkContextLeakTester.java:110)
			at org.apache.kafka.common.network.Selector.<init>(Selector.java:159)
			at org.apache.kafka.common.network.Selector.<init>(Selector.java:212)
			at org.apache.kafka.common.network.Selector.<init>(Selector.java:224)
			at org.apache.kafka.common.network.Selector.<init>(Selector.java:228)
			at org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:219)
			at org.apache.kafka.clients.ClientUtils.createNetworkClient(ClientUtils.java:160)
			at org.apache.kafka.clients.producer.KafkaProducer.newSender(KafkaProducer.java:514)
			at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:455)
			at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProducer.java:289)
			at org.apache.kafka.clients.producer.KafkaProducerTest.lambda$testDeliveryTimeoutAndLingerMsConfig$54(KafkaProducerTest.java:2427)
			at org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:71)
			at org.junit.jupiter.api.AssertDoesNotThrow.assertDoesNotThrow(AssertDoesNotThrow.java:58)
			at org.junit.jupiter.api.Assertions.assertDoesNotThrow(Assertions.java:3224)
			at org.apache.kafka.clients.producer.KafkaProducerTest.testDeliveryTimeoutAndLingerMsConfig(KafkaProducerTest.java:2427)
<snipped>
			... 60 more

It shows the exact stack trace of where the socket or selector was instantiated, and attributes the failure to the test which caused it. I found that this was sufficient to find and fix leaks in parts of the codebase I had never seen before, so it should be very helpful for people who already know their way around the test and the surrounding code.

Committer Checklist (excluded from commit message)

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

[github-actions]

This PR is being marked as stale since it has not had any activity in 90 days. If you would like to keep this PR alive, please ask a committer for review. If the PR has merge conflicts, please update it with the latest from trunk (or appropriate release branch)

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

[github-actions]

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

[divijvaidya]

@gharris1727 shall we try to push it across the finish line? We can begin to enable this, at least for tests which are passing this check now.

[github-actions]

This PR is being marked as stale since it has not had any activity in 90 days. If you
would like to keep this PR alive, please leave a comment asking for a review. If the PR has
merge conflicts, update it with the latest from the base branch.

If you are having difficulty finding a reviewer, please reach out on the [mailing list](https://kafka.apache.org/contact).

If this PR is no longer valid or desired, please feel free to close it. If no activity occurs in the next 30 days, it will be automatically closed.

[github-actions]

This PR has been closed since it has not had any activity in 120 days. If you feel like this
was a mistake, or you would like to continue working on it, please feel free to re-open the
PR and ask for a review.