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)
