[fix][io] Update Elasticsearch sink idle cnx timeout to 30s (#19377)

(cherry picked from commit 1481c74) (cherry picked from commit fd700da)
apache · Jan 31, 2023 · 5265cb8 · 5265cb8
1 parent 9d9c993
commit 5265cb8
Show file tree

Hide file tree

Showing 3 changed files with 4 additions and 5 deletions.
diff --git a/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java b/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchClient.java
@@ -188,7 +188,6 @@ public void afterBulk(long l, BulkRequest bulkRequest, Throwable throwable) {
         this.executorService.scheduleAtFixedRate(new Runnable() {
                                                      @Override
                                                      public void run() {
-                                                         configCallback.connectionManager.closeExpiredConnections();
                                                          configCallback.connectionManager.closeIdleConnections(
                                                                  config.getConnectionIdleTimeoutInMs(), TimeUnit.MILLISECONDS);
                                                      }

diff --git a/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java b/.../elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
@@ -193,10 +193,10 @@ public class ElasticSearchConfig implements Serializable {
 
     @FieldDoc(
             required = false,
-            defaultValue = "5",
-            help = "Idle connection timeout to prevent a read timeout."
+            defaultValue = "30000",
+            help = "Idle connection timeout to prevent a connection timeout due to inactivity."
     )
-    private int connectionIdleTimeoutInMs = 5;
+    private int connectionIdleTimeoutInMs = 30000;
 
     @FieldDoc(
             required = false,

diff --git a/...tic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java b/...tic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java
@@ -86,7 +86,7 @@ public final void defaultValueTest() throws IOException {
         assertEquals(config.isCompressionEnabled(), false);
         assertEquals(config.getConnectTimeoutInMs(), 5000L);
         assertEquals(config.getConnectionRequestTimeoutInMs(), 1000L);
-        assertEquals(config.getConnectionIdleTimeoutInMs(), 5L);
+        assertEquals(config.getConnectionIdleTimeoutInMs(), 30000L);
         assertEquals(config.getSocketTimeoutInMs(), 60000);
 
         assertEquals(config.isStripNulls(), true);