Greg Harris created KAFKA-15804:
-----------------------------------

             Summary: Broker leaks ServerSocketChannel when exception is thrown 
from ZkConfigManager during startup
                 Key: KAFKA-15804
                 URL: https://issues.apache.org/jira/browse/KAFKA-15804
             Project: Kafka
          Issue Type: Bug
          Components: core, Tiered-Storage, unit tests
    Affects Versions: 3.6.0
            Reporter: Greg Harris


This exception is thrown during the 
RemoteTopicCrudTest.testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic
 test in zk mode:
{noformat}
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

org.apache.kafka.storage.internals.log.LogConfig.validateRemoteStorageOnlyIfSystemEnabled(LogConfig.java:566)
kafka.log.LogManager.updateTopicConfig(LogManager.scala:956)
kafka.server.TopicConfigHandler.updateLogConfig(ConfigHandler.scala:73)
kafka.server.TopicConfigHandler.processConfigChanges(ConfigHandler.scala:94)
kafka.server.ZkConfigManager.$anonfun$startup$4(ZkConfigManager.scala:176)
kafka.server.ZkConfigManager.$anonfun$startup$4$adapted(ZkConfigManager.scala:175)
scala.collection.immutable.Map$Map2.foreach(Map.scala:360)
kafka.server.ZkConfigManager.$anonfun$startup$1(ZkConfigManager.scala:175)
kafka.server.ZkConfigManager.$anonfun$startup$1$adapted(ZkConfigManager.scala:166)
scala.collection.immutable.HashMap.foreach(HashMap.scala:1115)
kafka.server.ZkConfigManager.startup(ZkConfigManager.scala:166)
kafka.server.KafkaServer.startup(KafkaServer.scala:575)
kafka.integration.KafkaServerTestHarness.$anonfun$createBrokers$1(KafkaServerTestHarness.scala:356)
kafka.integration.KafkaServerTestHarness.$anonfun$createBrokers$1$adapted(KafkaServerTestHarness.scala:352)
scala.collection.immutable.List.foreach(List.scala:333)
kafka.integration.KafkaServerTestHarness.createBrokers(KafkaServerTestHarness.scala:352)
kafka.integration.KafkaServerTestHarness.recreateBrokers(KafkaServerTestHarness.scala:146)
kafka.admin.RemoteTopicCrudTest.$anonfun$testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic$1(RemoteTopicCrudTest.scala:319)
org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:53)
org.junit.jupiter.api.AssertThrows.assertThrows(AssertThrows.java:35)
org.junit.jupiter.api.Assertions.assertThrows(Assertions.java:3111)
kafka.admin.RemoteTopicCrudTest.testClusterWideDisablementOfTieredStorageWithEnabledTieredTopic(RemoteTopicCrudTest.scala:319){noformat}
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:
{noformat}
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){noformat}
And the associated DataPlaneAcceptor:
{noformat}
        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)
{noformat}
And two Processors:
{noformat}
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){noformat}
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.



--
This message was sent by Atlassian Jira
(v8.20.10#820010)

Reply via email to