diff --git a/src/main/java/org/sourcelab/kafka/connect/apiclient/Configuration.java b/src/main/java/org/sourcelab/kafka/connect/apiclient/Configuration.java
index c15e5e6..92af5e1 100644
--- a/src/main/java/org/sourcelab/kafka/connect/apiclient/Configuration.java
+++ b/src/main/java/org/sourcelab/kafka/connect/apiclient/Configuration.java
@@ -31,6 +31,7 @@ public final class Configuration {
 
     // Optional Connection options
     private int requestTimeoutInSeconds = 300;
+    private int connectionTimeToLiveInSeconds = 300;
 
     // Optional BasicAuth options
     private String basicAuthUsername = null;
@@ -164,6 +165,16 @@ public Configuration useRequestTimeoutInSeconds(final int requestTimeoutInSecond
         return this;
     }
 
+    /**
+     * Sets maximum time to live for persistent connections
+     * @param connectionTimeToLiveInSeconds How long before persistent connection will be interrupted, in seconds
+     * @return Configuration instance.
+     */
+    public Configuration useConnectionTimeToLiveInSeconds(final int connectionTimeToLiveInSeconds) {
+        this.connectionTimeToLiveInSeconds = connectionTimeToLiveInSeconds;
+        return this;
+    }
+
     public String getProxyHost() {
         return proxyHost;
     }
@@ -204,6 +215,10 @@ public int getRequestTimeoutInSeconds() {
         return requestTimeoutInSeconds;
     }
 
+    public int getConnectionTimeToLiveInSeconds() {
+        return connectionTimeToLiveInSeconds;
+    }
+
     public File getKeyStoreFile() {
         return keyStoreFile;
     }
@@ -224,7 +239,8 @@ public String getBasicAuthPassword() {
     public String toString() {
         final StringBuilder stringBuilder = new StringBuilder("Configuration{")
             .append("apiHost='").append(apiHost).append('\'')
-            .append(", requestTimeout='").append(requestTimeoutInSeconds).append('\'');
+            .append(", requestTimeout='").append(requestTimeoutInSeconds).append('\'')
+            .append("connectionTimeToLiveInSeconds='").append(connectionTimeToLiveInSeconds).append('\'');
         if (proxyHost != null) {
             stringBuilder
                 .append(", proxy='").append(proxyScheme).append("://");
diff --git a/src/main/java/org/sourcelab/kafka/connect/apiclient/rest/HttpClientRestClient.java b/src/main/java/org/sourcelab/kafka/connect/apiclient/rest/HttpClientRestClient.java
index f9e1764..4571dbe 100644
--- a/src/main/java/org/sourcelab/kafka/connect/apiclient/rest/HttpClientRestClient.java
+++ b/src/main/java/org/sourcelab/kafka/connect/apiclient/rest/HttpClientRestClient.java
@@ -110,7 +110,7 @@ public void init(final Configuration configuration) {
         final HttpClientBuilder clientBuilder = HttpClientBuilder.create();
         clientBuilder
             // Define timeout
-            .setConnectionTimeToLive(configuration.getRequestTimeoutInSeconds(), TimeUnit.SECONDS)
+            .setConnectionTimeToLive(configuration.getConnectionTimeToLiveInSeconds(), TimeUnit.SECONDS)
 
             // Define SSL Socket Factory instance.
             .setSSLSocketFactory(httpsContextBuilder.createSslSocketFactory());
@@ -118,6 +118,8 @@ public void init(final Configuration configuration) {
         // Define our RequestConfigBuilder
         final RequestConfig.Builder requestConfigBuilder = RequestConfig.custom();
 
+        requestConfigBuilder.setConnectTimeout(configuration.getRequestTimeoutInSeconds() * 1_000);
+
         // Define our Credentials Provider
         final CredentialsProvider credsProvider = new BasicCredentialsProvider();