Merge pull request #5 from mtfelisb/fix/httphost

Using HttpHost instead of host and port separately

Author: mtfelisb

pom.xml

@@ -6,7 +6,7 @@
 
     <groupId>com.mtfelisb</groupId>
     <artifactId>flink-connector-elasticsearch</artifactId>
-    <version>1.0.0</version>
+    <version>2.0.0</version>
     <packaging>jar</packaging>
 
     <licenses>

src/main/java/com/mtfelisb/flink/connectors/elasticsearch/sink/ElasticsearchSinkBuilder.java

@@ -21,6 +21,8 @@
 
 package com.mtfelisb.flink.connectors.elasticsearch.sink;
 
+import org.apache.http.HttpHost;
+
 import java.io.IOException;
 
 import static org.apache.flink.util.Preconditions.checkNotNull;
@@ -34,16 +36,10 @@
  */
 public class ElasticsearchSinkBuilder<T> {
     /**
-     * The host where the Elasticsearch cluster is reachable
-     *
-     */
-    private String host;
-
-    /**
-     * The port where the Elasticsearch cluster is reachable
+     * The HttpHost where the Elasticsearch cluster is reachable
      *
      */
-    private int port;
+    private HttpHost httpHost;
 
     /**
      * The username to authenticate the connection with the Elasticsearch cluster
@@ -70,29 +66,15 @@ public class ElasticsearchSinkBuilder<T> {
     private Emitter<T> emitter;
 
     /**
-     * setHost
-     * set the host where the Elasticsearch cluster is reachable
+     * setHttpHost
+     * set the HttpHost where the Elasticsearch cluster is reachable
      *
-     * @param host the host address
+     * @param httpHost the host address
      * @return this builder
      */
-    public ElasticsearchSinkBuilder<T> setHost(String host) {
-        checkNotNull(host);
-        checkState(host.length() > 0, "Host cannot be empty");
-        this.host = host;
-        return this;
-    }
-
-    /**
-     * setPort
-     * set the port where the Elasticsearch cluster is reachable
-     *
-     * @param port the port number
-     * @return
-     */
-    public ElasticsearchSinkBuilder<T> setPort(int port) {
-        // no need to checkNotNull() here since int can't be null
-        this.port = port;
+    public ElasticsearchSinkBuilder<T> setHttpHost(HttpHost httpHost) {
+        checkNotNull(httpHost);
+        this.httpHost = httpHost;
         return this;
     }
 
@@ -159,7 +141,7 @@ public ElasticsearchSink<T> build() {
         validate();
 
         return new ElasticsearchSink<T>(
-            new NetworkConfigFactory(host, port, username, password),
+            new NetworkConfigFactory(httpHost, username, password),
             emitter,
             threshold,
             new BulkRequestFactory()
@@ -172,8 +154,7 @@ public static <T> ElasticsearchSinkBuilder<T> builder() {
 
     private void validate() {
         this.setEmitter(this.emitter);
-        this.setHost(this.host);
-        this.setPort(this.port);
+        this.setHttpHost(this.httpHost);
         this.setThreshold(this.threshold);
     }
 }

src/main/java/com/mtfelisb/flink/connectors/elasticsearch/sink/NetworkConfigFactory.java

@@ -45,27 +45,24 @@
 public class NetworkConfigFactory implements INetworkConfigFactory {
     private final static Logger LOG = LogManager.getLogger(NetworkConfigFactory.class);
 
-    private final String host;
-
-    private final int port;
+    private final HttpHost httpHost;
 
     private final String username;
 
     private final String password;
 
-    public NetworkConfigFactory(String host, int port, String username, String password) {
-        this.host = checkNotNull(host);
-        this.port = checkNotNull(port);
+    public NetworkConfigFactory(HttpHost httpHost, String username, String password) {
+        this.httpHost = checkNotNull(httpHost);
         this.username = username;
         this.password = password;
     }
 
     @Override
     public ElasticsearchClient create() {
-        LOG.debug("Http client host {} and port {}", host, port);
+        LOG.debug("Http client hostname {}", httpHost.toHostString());
 
         // Create the low-level client
-        RestClient restClient = RestClient.builder(new HttpHost(host, port))
+        RestClient restClient = RestClient.builder(httpHost)
             .setHttpClientConfigCallback(httpClientBuilder ->
                 (username != null && password != null) ?
                     httpClientBuilder.setDefaultCredentialsProvider(getCredentials())

src/test/java/com/mtfelisb/flink/connectors/elasticsearch/sink/ElasticsearchSinkBuilderTest.java

@@ -21,6 +21,7 @@
 
 package com.mtfelisb.flink.connectors.elasticsearch.sink;
 
+import org.apache.http.HttpHost;
 import org.junit.jupiter.api.Test;
 
 import static org.junit.Assert.assertEquals;
@@ -53,8 +54,7 @@ public void sinkBuilderSetEmitter() {
         Throwable exception = assertThrows(
             NullPointerException.class,
             () -> ElasticsearchSinkBuilder.<String>builder()
-                .setPort(9200)
-                .setHost("localhost")
+                .setHttpHost(new HttpHost("localhost", 9200))
                 .setPassword("password")
                 .setUsername("username")
                 .setEmitter(null)
@@ -66,7 +66,7 @@ public void sinkBuilderSetEmitter() {
     }
 
     /**
-     * The host should be declared to create
+     * The httpHost should be declared to create
      * a valid ElasticsearchSink instance
      *
      */
@@ -75,30 +75,13 @@ public void sinkBuilderSetHost() {
         Throwable exception = assertThrows(
             NullPointerException.class,
             () -> ElasticsearchSinkBuilder.<String>builder()
-                .setHost(null)
+                .setHttpHost(null)
                 .build()
         );
 
         assertEquals(null, exception.getMessage());
     }
 
-    /**
-     * The host should be declared to create
-     * a valid ElasticsearchSink instance
-     *
-     */
-    @Test
-    public void sinkBuilderSetHostEmpty() {
-        Throwable exception = assertThrows(
-            IllegalStateException.class,
-            () -> ElasticsearchSinkBuilder.<String>builder()
-                .setHost("")
-                .build()
-        );
-
-        assertEquals("Host cannot be empty", exception.getMessage());
-    }
-
     /**
      * The username, if provided, cannot be null to create
      * a valid ElasticsearchSink instance

src/test/java/com/mtfelisb/flink/connectors/elasticsearch/sink/ElasticsearchSinkTest.java

@@ -60,8 +60,7 @@ public void indexingByThresholdReached() throws Exception {
 
         final ElasticsearchSink<DummyData> sink = ElasticsearchSinkBuilder.<DummyData>builder()
             .setThreshold(2L)
-            .setHost(ES_CONTAINER.getHost())
-            .setPort(ES_CONTAINER.getFirstMappedPort())
+            .setHttpHost(HttpHost.create(ES_CONTAINER.getHttpHostAddress()))
             .setEmitter(
                 (value, op, ctx) ->
                     (BulkOperation.Builder) op
@@ -102,8 +101,7 @@ public void indexingByCheckpoint() throws Exception {
 
         final ElasticsearchSink<DummyData> sink = ElasticsearchSinkBuilder.<DummyData>builder()
             .setThreshold(1000L)
-            .setHost(ES_CONTAINER.getHost())
-            .setPort(ES_CONTAINER.getFirstMappedPort())
+            .setHttpHost(HttpHost.create(ES_CONTAINER.getHttpHostAddress()))
             .setEmitter(
                 (value, op, ctx) ->
                     (BulkOperation.Builder) op

src/test/java/com/mtfelisb/flink/connectors/elasticsearch/sink/NetworkConfigFactoryTest.java

@@ -22,6 +22,7 @@
 package com.mtfelisb.flink.connectors.elasticsearch.sink;
 
 import co.elastic.clients.elasticsearch.ElasticsearchClient;
+import org.apache.http.HttpHost;
 import org.testcontainers.junit.jupiter.Testcontainers;
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertThrows;
@@ -38,7 +39,7 @@ public class NetworkConfigFactoryTest extends ElasticsearchSinkBaseITCase {
     public void requiredHost() {
         Throwable exception = assertThrows(
             NullPointerException.class,
-            () -> new NetworkConfigFactory(null, 9200, "username", "password"));
+            () -> new NetworkConfigFactory(null, "username", "password"));
 
         assertEquals(null, exception.getMessage());
     }
@@ -52,8 +53,7 @@ public void requiredHost() {
     @Test
     public void create() throws IOException {
         ElasticsearchClient esClient = new NetworkConfigFactory(
-            ES_CONTAINER.getHost(),
-            ES_CONTAINER.getFirstMappedPort(),
+            HttpHost.create(ES_CONTAINER.getHttpHostAddress()),
             null,
             null
         ).create();
@@ -70,8 +70,7 @@ public void create() throws IOException {
     @Test
     public void createWithUsernameAndPassword() throws IOException {
         ElasticsearchClient esClient = new NetworkConfigFactory(
-            ES_AUTH_CONTAINER.getHost(),
-            ES_AUTH_CONTAINER.getFirstMappedPort(),
+            HttpHost.create(ES_CONTAINER.getHttpHostAddress()),
             "elastic",
             "generic-pass"
         ).create();