Added protocol config to bypass certificates.

Author: Selcuk Holep

src/main/java/com/ibm/eventstreams/connect/elasticsink/ElasticSinkConnector.java

@@ -42,6 +42,11 @@ public class ElasticSinkConnector extends SinkConnector {
     public static final String CONFIG_DOCUMENTATION_ES_CONNECTION = "The hostname:port to connect to Elasticsearch.";
     public static final String CONFIG_DISPLAY_ES_CONNECTION = "hostname:port";
 
+    public static final String DEFAULT_ES_PROTOCOL = "http";
+    public static final String CONFIG_NAME_ES_PROTOCOL = "es.protocol";
+    public static final String CONFIG_DOCUMENTATION_ES_PROTOCOL = "The protocol to connect to Elasticsearch.";
+    public static final String CONFIG_DISPLAY_ES_PROTOCOL = "HTTP/ HTTPS";
+
     public static final String CONFIG_NAME_ES_USER_NAME = "es.user.name";
     public static final String CONFIG_DOCUMENTATION_ES_USER_NAME = "The user name for authenticating with Elasticsearch.";
     public static final String CONFIG_DISPLAY_ES_USER_NAME = "User name";
@@ -210,6 +215,10 @@ public ConfigDef config() {
                       CONFIG_DOCUMENTATION_ES_PASSWORD, CONFIG_GROUP_ES, 3, Width.MEDIUM,
                       CONFIG_DISPLAY_ES_PASSWORD);
 
+        config.define(CONFIG_NAME_ES_PROTOCOL, Type.STRING, DEFAULT_ES_PROTOCOL, Importance.MEDIUM,
+                CONFIG_DOCUMENTATION_ES_PROTOCOL, CONFIG_GROUP_ES, 4, Width.MEDIUM,
+                CONFIG_DISPLAY_ES_PROTOCOL);
+
         config.define(CONFIG_NAME_ES_TLS_KEYSTORE_LOCATION, Type.STRING, null, Importance.MEDIUM,
                       CONFIG_DOCUMENTATION_ES_TLS_KEYSTORE_LOCATION, CONFIG_GROUP_ES, 10, Width.MEDIUM,
                       CONFIG_DISPLAY_ES_TLS_KEYSTORE_LOCATION);

src/main/java/com/ibm/eventstreams/connect/elasticsink/ElasticWriter.java

@@ -69,7 +69,7 @@ public class ElasticWriter {
     private String trustStore;
     private String trustStorePassword;
 
-    private String protocol = "http";
+    private String protocol;
     private String proxyHost = null;
     private int proxyPort = 0;
 
@@ -107,6 +107,7 @@ public void configure(Map<String, String> props) throws ConnectException {
         keyStorePassword = props.get(ElasticSinkConnector.CONFIG_NAME_ES_TLS_KEYSTORE_PASSWORD);
         trustStore = props.get(ElasticSinkConnector.CONFIG_NAME_ES_TLS_TRUSTSTORE_LOCATION);
         trustStorePassword = props.get(ElasticSinkConnector.CONFIG_NAME_ES_TLS_TRUSTSTORE_PASSWORD);
+        protocol = props.getOrDefault(ElasticSinkConnector.CONFIG_NAME_ES_PROTOCOL,ElasticSinkConnector.DEFAULT_ES_PROTOCOL);
 
         // Jetty configuration for HTTP client behaviour
         proxyHost = props.get(ElasticSinkConnector.CONFIG_NAME_ES_HTTP_PROXY_HOST);