Camel http component does not close connections - Close_Wait

Question

Camel http component does not close connections properly?

Having below route I have observed that connections are being created on the server, but not terminated. After a while this is causing a problem

java.io.IOException: Too many open files

route:

from("seda:testSeda?concurrentConsumers=20")
    .setHeader("Connection", constant("Close"))
    .to("http://testServer/testFile.xml?authMethod=Basic&throwExceptionOnFailure=false&authUsername=user&authPassword=password")
    .to("file://abc")
.end();

connections are in Close_Wait state any ideas?

I am using camel-http lib in version 2.14

Answer 1

You can override default HttpClient used by Apache Camel and define a custom Keep Alive Strategy.

https://howtodoinjava.com/spring-boot2/resttemplate/resttemplate-httpclient-java-config/

The code bellow resolved my issue in production:

@Configuration public class AppConfiguration {

@Autowired
private PoolingHttpClientConnectionManager poolingConnectionManager;
@Autowired
private ConnectionKeepAliveStrategy connectionKeepAliveStrategy;
@Autowired
private SSLConnectionSocketFactory sslContext;

@Bean
CamelContextConfiguration contextConfiguration() {
    return new CamelContextConfiguration() {
        @Override
        public void beforeApplicationStart(CamelContext context) {
            HttpComponent httpComponent = context.getComponent("https4", HttpComponent.class);

            httpComponent.setHttpClientConfigurer(new HttpClientConfigurer() {
                @Override
                public void configureHttpClient(HttpClientBuilder builder) {

                    builder.setSSLSocketFactory(sslContext);

                    RegistryBuilder.<ConnectionSocketFactory>create().register("https", sslContext).build();

                    builder.setConnectionManager(poolingConnectionManager);
                    builder.setKeepAliveStrategy(connectionKeepAliveStrategy);
                }
            });
        }

        @Override
        public void afterApplicationStart(CamelContext arg0) {

        }

    };
}

}

@Configuration public class HttpClientConfig {

private static final int DEFAULT_KEEP_ALIVE_TIME_MILLIS = 20 * 1000;
private static final int CLOSE_IDLE_CONNECTION_WAIT_TIME_SECS = 30;

@Value("${pathCertificado}")
private String pathCertificado;

private Logger logger = LoggerFactory.getLogger(HttpClientConfig.class);

@Bean
public PoolingHttpClientConnectionManager poolingConnectionManager() {
    PoolingHttpClientConnectionManager connectionManager = new PoolingHttpClientConnectionManager();
    connectionManager.setMaxTotal(200);
    connectionManager.setDefaultMaxPerRoute(20);

    return connectionManager;
}

@Bean
public CloseableHttpClient httpClient() {
    RequestConfig config = RequestConfig.custom().setConnectionRequestTimeout(5000).setConnectTimeout(5000)
            .setSocketTimeout(15000).build();

    return HttpClientBuilder.create().setSSLSocketFactory(this.getSSLContext())
            .setConnectionManager(this.poolingConnectionManager()).setDefaultRequestConfig(config)
            .setKeepAliveStrategy(this.connectionKeepAliveStrategy()).build();

}

@Bean
public ConnectionKeepAliveStrategy connectionKeepAliveStrategy() {
    return new ConnectionKeepAliveStrategy() {
        @Override
        public long getKeepAliveDuration(HttpResponse response, HttpContext context) {
            HeaderElementIterator it = new BasicHeaderElementIterator(
                    response.headerIterator(HTTP.CONN_KEEP_ALIVE));
            while (it.hasNext()) {
                HeaderElement he = it.nextElement();
                String param = he.getName();
                String value = he.getValue();

                if (value != null && param.equalsIgnoreCase("timeout")) {
                    return Long.parseLong(value) * 1000;
                }
            }
            return DEFAULT_KEEP_ALIVE_TIME_MILLIS;
        }
    };
}

@Bean
public Runnable idleConnectionMonitor(final PoolingHttpClientConnectionManager connectionManager) {
    return new Runnable() {
        @Override
        @Scheduled(fixedDelay = 10000)
        public void run() {
            if (connectionManager != null) {
                connectionManager.closeExpiredConnections();
                connectionManager.closeIdleConnections(CLOSE_IDLE_CONNECTION_WAIT_TIME_SECS, TimeUnit.SECONDS);
            }
        }
    };
}

@Bean
public SSLConnectionSocketFactory getSSLContext() {
    try {
        KeyStore keyStore = KeyStore.getInstance(KeyStore.getDefaultType());

        try (FileInputStream jksFile = new FileInputStream(this.pathCertificado)) {
            keyStore.load(jksFile, "xxxxxx".toCharArray());
        }

        TrustStrategy acceptingTrustStrategy = (X509Certificate[] chain, String authType) -> true;

        SSLContext sslContext = SSLContexts.custom().loadTrustMaterial(keyStore, acceptingTrustStrategy).build();

        return new SSLConnectionSocketFactory(sslContext);
    } catch (Exception e) {
        logger.error("Keystore load failed: " + this.pathCertificado, e);

        return null;
    }
}

}