From 3a405f11e55a8b5fd787c73fb59778e705012df9 Mon Sep 17 00:00:00 2001 From: xiaolongran Date: Mon, 27 Apr 2026 15:25:49 +0800 Subject: [PATCH 1/2] [Improve][Broker]Support broker config TCP keepalive options Signed-off-by: xiaolongran --- .../pulsar/broker/ServiceConfiguration.java | 25 +++++++++++++++++++ .../broker/BookKeeperClientFactoryImpl.java | 9 +++++++ .../BookKeeperClientFactoryImplTest.java | 22 ++++++++++++++++ 3 files changed, 56 insertions(+) diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index b03af49c56eef..f300b92431863 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -357,6 +357,31 @@ public class ServiceConfiguration implements PulsarConfiguration { + "(0 to disable limiting)") private int maxHttpServerConnections = 2048; + @FieldContext( + category = CATEGORY_SERVER, + doc = "TCP keepalive time in seconds. This is the duration between the last data packet sent " + + "(simple ACKs are not considered data) and the first keepalive probe. " + + "Default is 7200 seconds (2 hours). " + + "If set to 0, the system default value will be used." + ) + private int tcpKeepAliveTimeSeconds = 0; + + @FieldContext( + category = CATEGORY_SERVER, + doc = "TCP keepalive interval in seconds. This is the duration between subsequent keepalive probes, " + + "if no acknowledgement is received from the peer. Default is 75 seconds. " + + "If set to 0, the system default value will be used." + ) + private int tcpKeepAliveIntervalSeconds = 0; + + @FieldContext( + category = CATEGORY_SERVER, + doc = "TCP keepalive probe count. This is the number of unacknowledged probes to send before considering " + + "the connection dead and notifying the application layer. Default is 9 probes. " + + "If set to 0, the system default value will be used." + ) + private int tcpKeepAliveProbeCount = 0; + @FieldContext(category = CATEGORY_SERVER, doc = "Gzip compression is enabled by default. Specific paths can be excluded from compression.\n" + "There are 2 syntaxes supported, Servlet url-pattern based, and Regex based.\n" diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java index f528ec3a7b63c..8fbfb8064e3b6 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java @@ -138,6 +138,15 @@ ClientConfiguration createBkClientConfiguration(MetadataStoreExtended store, Ser bkConf.setDiskWeightBasedPlacementEnabled(conf.isBookkeeperDiskWeightBasedPlacementEnabled()); bkConf.setMetadataServiceUri(conf.getBookkeeperMetadataStoreUrl()); bkConf.setLimitStatsLogging(conf.isBookkeeperClientLimitStatsLogging()); + if (conf.getTcpKeepAliveTimeSeconds() > 0) { + bkConf.setTcpKeepIdle(conf.getTcpKeepAliveTimeSeconds()); + } + if (conf.getTcpKeepAliveIntervalSeconds() > 0) { + bkConf.setTcpKeepIntvl(conf.getTcpKeepAliveIntervalSeconds()); + } + if (conf.getTcpKeepAliveProbeCount() > 0) { + bkConf.setTcpKeepCnt(conf.getTcpKeepAliveProbeCount()); + } if (!conf.isBookkeeperMetadataStoreSeparated()) { // If we're connecting to the same metadata service, with same config, then diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java index fdf8dea0f0413..d5ba87fca76fa 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java @@ -341,4 +341,26 @@ public void testBookKeeperLimitStatsLoggingConfiguration() throws Exception { (ClientConfiguration) FieldUtils.readField(builder, "conf", true); assertFalse(clientConfiguration.getLimitStatsLogging()); } + + @Test + public void testSetTcpKeepAliveConfiguration() { + BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl(); + ServiceConfiguration conf = new ServiceConfiguration(); + + // test default value + ClientConfiguration bkConf = factory.createBkClientConfiguration(mock(MetadataStoreExtended.class), conf); + assertEquals(bkConf.getTcpKeepIdle(), -1); + assertEquals(bkConf.getTcpKeepIntvl(), -1); + assertEquals(bkConf.getTcpKeepCnt(), -1); + + // test setting TCP keep-alive parameters + conf.setTcpKeepAliveTimeSeconds(60); + conf.setTcpKeepAliveIntervalSeconds(10); + conf.setTcpKeepAliveProbeCount(5); + + bkConf = factory.createBkClientConfiguration(mock(MetadataStoreExtended.class), conf); + assertEquals(bkConf.getTcpKeepIdle(), 60); + assertEquals(bkConf.getTcpKeepIntvl(), 10); + assertEquals(bkConf.getTcpKeepCnt(), 5); + } } From 6e7bd3c4fe11601b670efa26ff63c05fc9b49ae7 Mon Sep 17 00:00:00 2001 From: xiaolongran Date: Mon, 27 Apr 2026 19:56:07 +0800 Subject: [PATCH 2/2] fix comments Signed-off-by: xiaolongran --- conf/broker.conf | 22 ++++++++++++++++ .../pulsar/broker/ServiceConfiguration.java | 25 ------------------- .../broker/BookKeeperClientFactoryImpl.java | 9 ------- .../BookKeeperClientFactoryImplTest.java | 22 ---------------- 4 files changed, 22 insertions(+), 56 deletions(-) diff --git a/conf/broker.conf b/conf/broker.conf index 4d6bb9162bf62..d6086f1e389f0 100644 --- a/conf/broker.conf +++ b/conf/broker.conf @@ -1256,6 +1256,28 @@ managedLedgerDefaultAckQuorum=2 # the BookkeeperPackagesStorage bookkeeper client), except the distributed log bookkeeper client. # The dlog bookkeeper client is configured in the functions worker configuration file. +# TCP keep-alive settings for the BookKeeper client connections from the broker to bookies. +# These options were introduced in BookKeeper 4.17.3 (see apache/bookkeeper#4683) and are +# forwarded to the BookKeeper client via the generic "bookkeeper_" prefix mechanism above. +# A value of -1 (BookKeeper default) defers to the operating system's TCP keep-alive settings. +# Uncomment and adjust to enable finer-grained keep-alive probing from the broker. +# +# Note: TCP keep-alive is a two-sided mechanism. Tuning these on the broker (BookKeeper +# client) side only affects connections initiated by the broker. To complete the picture +# you should also tune the bookie node's OS-level keep-alive settings via sysctl, e.g. +# net.ipv4.tcp_keepalive_time / tcp_keepalive_intvl / tcp_keepalive_probes, since the +# BookKeeper server (bookie) currently does not expose application-level keep-alive knobs. + +# Idle time (in seconds) before the first TCP keep-alive probe is sent on an otherwise idle +# BookKeeper client connection. +#bookkeeper_tcpKeepIdle=300 +# Interval (in seconds) between subsequent TCP keep-alive probes when no acknowledgement is +# received from the bookie. +#bookkeeper_tcpKeepIntvl=60 +# Number of unacknowledged TCP keep-alive probes before the connection is considered dead +# and closed. +#bookkeeper_tcpKeepCnt=5 + # How frequently to flush the cursor positions that were accumulated due to rate limiting. (seconds). # Default is 60 seconds managedLedgerCursorPositionFlushSeconds=60 diff --git a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java index f300b92431863..b03af49c56eef 100644 --- a/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java +++ b/pulsar-broker-common/src/main/java/org/apache/pulsar/broker/ServiceConfiguration.java @@ -357,31 +357,6 @@ public class ServiceConfiguration implements PulsarConfiguration { + "(0 to disable limiting)") private int maxHttpServerConnections = 2048; - @FieldContext( - category = CATEGORY_SERVER, - doc = "TCP keepalive time in seconds. This is the duration between the last data packet sent " - + "(simple ACKs are not considered data) and the first keepalive probe. " - + "Default is 7200 seconds (2 hours). " - + "If set to 0, the system default value will be used." - ) - private int tcpKeepAliveTimeSeconds = 0; - - @FieldContext( - category = CATEGORY_SERVER, - doc = "TCP keepalive interval in seconds. This is the duration between subsequent keepalive probes, " - + "if no acknowledgement is received from the peer. Default is 75 seconds. " - + "If set to 0, the system default value will be used." - ) - private int tcpKeepAliveIntervalSeconds = 0; - - @FieldContext( - category = CATEGORY_SERVER, - doc = "TCP keepalive probe count. This is the number of unacknowledged probes to send before considering " - + "the connection dead and notifying the application layer. Default is 9 probes. " - + "If set to 0, the system default value will be used." - ) - private int tcpKeepAliveProbeCount = 0; - @FieldContext(category = CATEGORY_SERVER, doc = "Gzip compression is enabled by default. Specific paths can be excluded from compression.\n" + "There are 2 syntaxes supported, Servlet url-pattern based, and Regex based.\n" diff --git a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java index 8fbfb8064e3b6..f528ec3a7b63c 100644 --- a/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java +++ b/pulsar-broker/src/main/java/org/apache/pulsar/broker/BookKeeperClientFactoryImpl.java @@ -138,15 +138,6 @@ ClientConfiguration createBkClientConfiguration(MetadataStoreExtended store, Ser bkConf.setDiskWeightBasedPlacementEnabled(conf.isBookkeeperDiskWeightBasedPlacementEnabled()); bkConf.setMetadataServiceUri(conf.getBookkeeperMetadataStoreUrl()); bkConf.setLimitStatsLogging(conf.isBookkeeperClientLimitStatsLogging()); - if (conf.getTcpKeepAliveTimeSeconds() > 0) { - bkConf.setTcpKeepIdle(conf.getTcpKeepAliveTimeSeconds()); - } - if (conf.getTcpKeepAliveIntervalSeconds() > 0) { - bkConf.setTcpKeepIntvl(conf.getTcpKeepAliveIntervalSeconds()); - } - if (conf.getTcpKeepAliveProbeCount() > 0) { - bkConf.setTcpKeepCnt(conf.getTcpKeepAliveProbeCount()); - } if (!conf.isBookkeeperMetadataStoreSeparated()) { // If we're connecting to the same metadata service, with same config, then diff --git a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java index d5ba87fca76fa..fdf8dea0f0413 100644 --- a/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java +++ b/pulsar-broker/src/test/java/org/apache/pulsar/broker/BookKeeperClientFactoryImplTest.java @@ -341,26 +341,4 @@ public void testBookKeeperLimitStatsLoggingConfiguration() throws Exception { (ClientConfiguration) FieldUtils.readField(builder, "conf", true); assertFalse(clientConfiguration.getLimitStatsLogging()); } - - @Test - public void testSetTcpKeepAliveConfiguration() { - BookKeeperClientFactoryImpl factory = new BookKeeperClientFactoryImpl(); - ServiceConfiguration conf = new ServiceConfiguration(); - - // test default value - ClientConfiguration bkConf = factory.createBkClientConfiguration(mock(MetadataStoreExtended.class), conf); - assertEquals(bkConf.getTcpKeepIdle(), -1); - assertEquals(bkConf.getTcpKeepIntvl(), -1); - assertEquals(bkConf.getTcpKeepCnt(), -1); - - // test setting TCP keep-alive parameters - conf.setTcpKeepAliveTimeSeconds(60); - conf.setTcpKeepAliveIntervalSeconds(10); - conf.setTcpKeepAliveProbeCount(5); - - bkConf = factory.createBkClientConfiguration(mock(MetadataStoreExtended.class), conf); - assertEquals(bkConf.getTcpKeepIdle(), 60); - assertEquals(bkConf.getTcpKeepIntvl(), 10); - assertEquals(bkConf.getTcpKeepCnt(), 5); - } }