feat(exporter): define remote connection timeout (#158)

* define the connection timeout to the remote cluster in the exporter config

README.md

@@ -145,6 +145,30 @@ The values can be overridden by environment variables with the same name and a `
 
 By default, the exporter creates an in-memory Hazelcast instance and publishes the records to it. But it can also be configured to export the records to a remote/external Hazecast instance by setting the argument `remoteAddress` or the environment variable `ZEEBE_HAZELCAST_REMOTE_ADDRESS` to the address of the remote Hazelcast instance.
 
+* the remote Hazelcast address
+* the remote Hazelcast cluster name
+* the connection timeout
+
+Default values:
+
+```
+zeebe:
+  broker:
+    exporters:
+      hazelcast:
+        className: io.zeebe.hazelcast.exporter.HazelcastExporter
+        jarPath: exporters/zeebe-hazelcast-exporter.jar
+	args:
+	      # remote Hazelcast address
+    	  remoteAddress = 127.0.0.1:5702
+
+    	  # Hazelcast cluster name
+    	  clusterName = "dev"
+
+    	  # connection timeout
+    	  remoteConnectionTimeout = "PT30S"
+```
+
 <details>
   <summary>Full docker-compose.yml with external Hazelcast</summary>
   <p>

exporter/src/main/java/io/zeebe/hazelcast/exporter/ExporterConfiguration.java

@@ -8,8 +8,6 @@ public class ExporterConfiguration {
 
   private int port = 5701;
 
-  private String remoteAddress;
-
   private String clusterName = "dev";
 
   private String name = "zeebe";
@@ -22,6 +20,9 @@ public class ExporterConfiguration {
   private String enabledValueTypes = "";
   private String enabledRecordTypes = "";
 
+  private String remoteAddress;
+  private String remoteConnectionTimeout = "PT30S";
+
   public int getPort() {
     return getEnv("PORT").map(Integer::parseInt).orElse(port);
   }
@@ -60,6 +61,10 @@ public String getClusterName() {
     return getEnv("CLUSTER_NAME").orElse(clusterName);
   }
 
+  public String getRemoteConnectionTimeout() {
+    return getEnv("REMOTE_CONNECTION_TIMEOUT").orElse(remoteConnectionTimeout);
+  }
+
   private Optional<String> getEnv(String name) {
     return Optional.ofNullable(System.getenv(ENV_PREFIX + name));
   }
@@ -84,6 +89,8 @@ public String toString() {
         + timeToLiveInSeconds
         + ", format="
         + format
+        + ", remoteConnectionTimeout="
+        + remoteConnectionTimeout
         + "]";
   }
 }

exporter/src/main/java/io/zeebe/hazelcast/exporter/HazelcastExporter.java

@@ -15,6 +15,7 @@
 import io.zeebe.exporter.proto.Schema;
 import org.slf4j.Logger;
 
+import java.time.Duration;
 import java.util.function.Function;
 
 public class HazelcastExporter implements Exporter {
@@ -116,11 +117,15 @@ private HazelcastInstance connectToHazelcast(String remoteAddress) {
 
     final var clientConfig = new ClientConfig();
     clientConfig.setProperty("hazelcast.logging.type", "slf4j");
+    clientConfig.setClusterName(clusterName);
 
     final var networkConfig = clientConfig.getNetworkConfig();
     networkConfig.addAddress(remoteAddress);
 
-    clientConfig.setClusterName(clusterName);
+    final var connectionRetryConfig =
+        clientConfig.getConnectionStrategyConfig().getConnectionRetryConfig();
+    final var connectionTimeout = Duration.parse(this.config.getRemoteConnectionTimeout());
+    connectionRetryConfig.setClusterConnectTimeoutMillis(connectionTimeout.toMillis());
 
     logger.info(
         "Connecting to remote Hazelcast instance [address: {}, cluster-name: {}]",