[KAFKA-15804] Broker leaks ServerSocketChannel when exception is thrown from ZkConfigManager during startup

• Type: Bug
• Status: Resolved
• Priority: Minor
• Resolution: Fixed
• Affects Version/s: 3.6.0
• Fix Version/s: 3.8.0
• Component/s: core, unit tests
• Assignee: Greg Harris
• Reporter: Greg Harris
• Created: 09/Nov/23 23:27
• Updated: 05/Aug/24 18:08
• Resolved: 10/May/24 20:57

Description

This exception is thrown during the RemoteTopicCrudTest.testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic test in zk mode:

org.apache.kafka.common.config.ConfigException: You have to delete all topics with the property remote.storage.enable=true before disabling tiered storage cluster-wide
        at org.apache.kafka.storage.internals.log.LogConfig.validateRemoteStorageOnlyIfSystemEnabled(LogConfig.java:566)
        at kafka.log.LogManager.updateTopicConfig(LogManager.scala:956)
        at kafka.server.TopicConfigHandler.updateLogConfig(ConfigHandler.scala:73)
        at kafka.server.TopicConfigHandler.processConfigChanges(ConfigHandler.scala:94)
        at kafka.server.ZkConfigManager.$anonfun$startup$4(ZkConfigManager.scala:176)
        at kafka.server.ZkConfigManager.$anonfun$startup$4$adapted(ZkConfigManager.scala:175)
        at scala.collection.immutable.Map$Map2.foreach(Map.scala:360)
        at kafka.server.ZkConfigManager.$anonfun$startup$1(ZkConfigManager.scala:175)
        at kafka.server.ZkConfigManager.$anonfun$startup$1$adapted(ZkConfigManager.scala:166)
        at scala.collection.immutable.HashMap.foreach(HashMap.scala:1115)
        at kafka.server.ZkConfigManager.startup(ZkConfigManager.scala:166)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:575)
        at kafka.integration.KafkaServerTestHarness.$anonfun$createBrokers$1(KafkaServerTestHarness.scala:356)
        at kafka.integration.KafkaServerTestHarness.$anonfun$createBrokers$1$adapted(KafkaServerTestHarness.scala:352)
        at scala.collection.immutable.List.foreach(List.scala:333)
        at kafka.integration.KafkaServerTestHarness.createBrokers(KafkaServerTestHarness.scala:352)
        at kafka.integration.KafkaServerTestHarness.recreateBrokers(KafkaServerTestHarness.scala:146)
        at kafka.admin.RemoteTopicCrudTest.$anonfun$testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic$1(RemoteTopicCrudTest.scala:319)
        at org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:53)
        at org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:35)
        at org.junit.jupiter.api.Assertions.assertThrows(Assertions.java:3111)
        at kafka.admin.RemoteTopicCrudTest.testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic(RemoteTopicCrudTest.scala:319)

This leak only occurs for this one test in the RemoteTopicCrudTest; all other tests including the kraft-mode version do not exhibit a leaked socket.

Here is where the ServerSocket is instantiated:

        at java.base/java.nio.channels.ServerSocketChannel.open(ServerSocketChannel.java:113)
        at kafka.network.Acceptor.openServerSocket(SocketServer.scala:724)
        at kafka.network.Acceptor.<init>(SocketServer.scala:608)
        at kafka.network.DataPlaneAcceptor.<init>(SocketServer.scala:454)
        at kafka.network.SocketServer.createDataPlaneAcceptor(SocketServer.scala:270)
        at kafka.network.SocketServer.createDataPlaneAcceptorAndProcessors(SocketServer.scala:249)
        at kafka.network.SocketServer.$anonfun$new$31(SocketServer.scala:175)
        at kafka.network.SocketServer.$anonfun$new$31$adapted(SocketServer.scala:175)
        at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:576)
        at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:574)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:933)
        at kafka.network.SocketServer.<init>(SocketServer.scala:175)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:344)
        at kafka.integration.KafkaServerTestHarness.$anonfun$createBrokers$1(KafkaServerTestHarness.scala:356)
        at kafka.integration.KafkaServerTestHarness.$anonfun$createBrokers$1$adapted(KafkaServerTestHarness.scala:352)
        at scala.collection.immutable.List.foreach(List.scala:333)
        at kafka.integration.KafkaServerTestHarness.createBrokers(KafkaServerTestHarness.scala:352)
        at kafka.integration.KafkaServerTestHarness.recreateBrokers(KafkaServerTestHarness.scala:146)
        at kafka.admin.RemoteTopicCrudTest.$anonfun$testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic$1(RemoteTopicCrudTest.scala:319)
        at org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:53)
        at org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:35)
        at org.junit.jupiter.api.Assertions.assertThrows(Assertions.java:3111)
        at kafka.admin.RemoteTopicCrudTest.testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic(RemoteTopicCrudTest.scala:319)

And the associated DataPlaneAcceptor:

         at java.base/java.nio.channels.Selector.open(Selector.java:295)
         at kafka.network.Acceptor.<init>(SocketServer.scala:598)
         at kafka.network.DataPlaneAcceptor.<init>(SocketServer.scala:454)
         at kafka.network.SocketServer.createDataPlaneAcceptor(SocketServer.scala:270)
         at kafka.network.SocketServer.createDataPlaneAcceptorAndProcessors(SocketServer.scala:249)
         at kafka.network.SocketServer.$anonfun$new$31(SocketServer.scala:175)
         at kafka.network.SocketServer.$anonfun$new$31$adapted(SocketServer.scala:175)
         at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:576)
         at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:574)
         at scala.collection.AbstractIterable.foreach(Iterable.scala:933)
         at kafka.network.SocketServer.<init>(SocketServer.scala:175)
         at kafka.server.KafkaServer.startup(KafkaServer.scala:344)

And two Processors:

        at java.base/java.nio.channels.Selector.open(Selector.java:295)
        at org.apache.kafka.common.network.Selector.<init>(Selector.java:159)
        at kafka.network.Processor.createSelector(SocketServer.scala:995)
        at kafka.network.Processor.<init>(SocketServer.scala:973)
        at kafka.network.Acceptor.newProcessor(SocketServer.scala:879)
        at kafka.network.Acceptor.$anonfun$addProcessors$1(SocketServer.scala:849)
        at scala.collection.immutable.Range.foreach$mVc$sp(Range.scala:190)
        at kafka.network.Acceptor.addProcessors(SocketServer.scala:848)
        at kafka.network.DataPlaneAcceptor.configure(SocketServer.scala:523)
        at kafka.network.SocketServer.createDataPlaneAcceptorAndProcessors(SocketServer.scala:251)
        at kafka.network.SocketServer.$anonfun$new$31(SocketServer.scala:175)
        at kafka.network.SocketServer.$anonfun$new$31$adapted(SocketServer.scala:175)
        at scala.collection.IterableOnceOps.foreach(IterableOnce.scala:576)
        at scala.collection.IterableOnceOps.foreach$(IterableOnce.scala:574)
        at scala.collection.AbstractIterable.foreach(Iterable.scala:933)
        at kafka.network.SocketServer.<init>(SocketServer.scala:175)
        at kafka.server.KafkaServer.startup(KafkaServer.scala:344)

I found this while investigating the causes of recent build failures where Gradle is unable to connect to the test runners. I don't believe this bug has any impact on end-users, but may have an impact on users running an embedded kafka in a test environment with a long-lived JVM.
I have seen other tests leaking sockets/leaving files open, so the RemoteTopicCrudTest is not unique. It just happens to be the first test i've looked into.

Issue Links

• causes: KAFKA-17268 Fix NPE when closing a non-started acceptor
• Discovered while testing: KAFKA-15845 Add Junit5 test extension which detects leaked Kafka clients and servers
• links to: GitHub Pull Request #14729