Merge remote-tracking branch 'origin/11.0.x' into HEAD

Author: avocader

src/test/java/io/confluent/connect/elasticsearch/ElasticsearchClientTest.java

@@ -573,7 +573,7 @@ public void testSsl() throws Exception {
     container = ElasticsearchContainer.fromSystemProperties().withSslEnabled(true);
     container.start();
 
-    String address = container.getConnectionUrl().replace(container.getContainerIpAddress(), container.hostMachineIpAddress());
+    String address = container.getConnectionUrl(false);
     props.put(CONNECTION_URL_CONFIG, address);
     props.put(CONNECTION_USERNAME_CONFIG, ELASTIC_SUPERUSER_NAME);
     props.put(CONNECTION_PASSWORD_CONFIG, ELASTIC_SUPERUSER_PASSWORD);

src/test/java/io/confluent/connect/elasticsearch/helper/ElasticsearchContainer.java

@@ -15,6 +15,7 @@
 
 package io.confluent.connect.elasticsearch.helper;
 
+import org.apache.kafka.common.config.SslConfigs;
 import org.elasticsearch.client.security.user.User;
 import org.elasticsearch.client.security.user.privileges.Role;
 import org.slf4j.Logger;
@@ -41,10 +42,13 @@
 import java.util.Map;
 
 import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig;
+import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SecurityProtocol;
 
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_PASSWORD_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_USERNAME_CONFIG;
+import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SECURITY_PROTOCOL_CONFIG;
+import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SSL_CONFIG_PREFIX;
 
 /**
  * A specialized TestContainer container for testing Elasticsearch, optionally with SSL support.
@@ -154,16 +158,32 @@ public ElasticsearchContainer(String imageName) {
   @Override
   public void start() {
     super.start();
+    String address;
     if (isBasicAuthEnabled()) {
       Map<String, String> props = new HashMap<>();
       props.put(CONNECTION_USERNAME_CONFIG, ELASTIC_SUPERUSER_NAME);
       props.put(CONNECTION_PASSWORD_CONFIG, ELASTIC_SUPERUSER_PASSWORD);
-      props.put(CONNECTION_URL_CONFIG, getConnectionUrl());
+      if (isSslEnabled()) {
+        addSslProps(props);
+        address = this.getConnectionUrl(false);
+      } else {
+        address = this.getConnectionUrl();
+      }
+      props.put(CONNECTION_URL_CONFIG, address);
       ElasticsearchHelperClient helperClient = getHelperClient(props);
       createUsersAndRoles(helperClient);
     }
   }
 
+  public void addSslProps(Map<String, String> props) {
+    props.put(SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name());
+    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, this.getKeystorePath());
+    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, this.getKeystorePassword());
+    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, this.getTruststorePath());
+    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, this.getTruststorePassword());
+    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_KEY_PASSWORD_CONFIG, this.getKeyPassword());
+  }
+
   private void createUsersAndRoles(ElasticsearchHelperClient helperClient ) {
     try {
       for (Role role: this.rolesToCreate) {
@@ -371,19 +391,28 @@ public String hostMachineIpAddress() {
     }
   }
 
+  /**
+   * @see ElasticsearchContainer#getConnectionUrl(boolean)
+   */
+  public String getConnectionUrl() {
+    return getConnectionUrl(true);
+  }
+
   /**
    * Get the Elasticsearch connection URL.
    *
    * <p>This can only be called once the container is started.
    *
+   * @param useContainerIpAddress use container IP if true, host machine's IP otherwise
+   *
    * @return the connection URL; never null
    */
-  public String getConnectionUrl() {
+  public String getConnectionUrl(boolean useContainerIpAddress) {
     String protocol = isSslEnabled() ? "https" : "http";
     return String.format(
         "%s://%s:%d",
         protocol,
-        getContainerIpAddress(),
+        useContainerIpAddress ? getContainerIpAddress() : hostMachineIpAddress(),
         getMappedPort(ELASTICSEARCH_DEFAULT_PORT)
     );
   }
@@ -524,7 +553,7 @@ public ElasticsearchHelperClient getHelperClient(Map<String, String> props) {
     superUserProps.put(CONNECTION_USERNAME_CONFIG, ELASTIC_SUPERUSER_NAME);
     superUserProps.put(CONNECTION_PASSWORD_CONFIG, ELASTIC_SUPERUSER_PASSWORD);
     ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(superUserProps);
-    ElasticsearchHelperClient client = new ElasticsearchHelperClient(this.getConnectionUrl(), config);
+    ElasticsearchHelperClient client = new ElasticsearchHelperClient(props.get(CONNECTION_URL_CONFIG), config);
     return client;
   }
 }

src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchConnectorIT.java

@@ -31,7 +31,7 @@
 import io.confluent.connect.elasticsearch.helper.ElasticsearchContainer;
 
 import org.apache.kafka.connect.storage.StringConverter;
-import org.apache.kafka.test.IntegrationTest;
+import io.confluent.common.utils.IntegrationTest;
 import org.elasticsearch.client.security.user.User;
 import org.elasticsearch.client.security.user.privileges.Role;
 import org.elasticsearch.search.SearchHit;

src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchConnectorKerberosIT.java

@@ -13,10 +13,13 @@
 import java.util.Properties;
 import org.apache.hadoop.minikdc.MiniKdc;
 import org.apache.kafka.connect.errors.ConnectException;
+import io.confluent.common.utils.IntegrationTest;
 import org.junit.AfterClass;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
+@Category(IntegrationTest.class)
 public class ElasticsearchConnectorKerberosIT extends ElasticsearchConnectorBaseIT {
 
   private static File baseDir;

src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchConnectorKerberosWithSslIT.java

@@ -7,9 +7,12 @@
 import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SecurityProtocol;
 import io.confluent.connect.elasticsearch.helper.ElasticsearchContainer;
 import org.apache.kafka.common.config.SslConfigs;
+import io.confluent.common.utils.IntegrationTest;
 import org.junit.BeforeClass;
 import org.junit.Test;
+import org.junit.experimental.categories.Category;
 
+@Category(IntegrationTest.class)
 public class ElasticsearchConnectorKerberosWithSslIT extends ElasticsearchConnectorKerberosIT{
 
   @BeforeClass
@@ -33,10 +36,7 @@ public void testKerberos() {
   @Test
   public void testKerberosWithSsl() throws Exception {
     // Use IP address here because that's what the certificates allow
-    String address = container.getConnectionUrl().replace(
-        container.getContainerIpAddress(),
-        container.hostMachineIpAddress()
-    );
+    String address = container.getConnectionUrl(false);
 
     props.put(CONNECTION_URL_CONFIG, address);
     props.put(SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name());

src/test/java/io/confluent/connect/elasticsearch/integration/ElasticsearchConnectorSslIT.java

@@ -16,7 +16,6 @@
 package io.confluent.connect.elasticsearch.integration;
 
 import io.confluent.common.utils.IntegrationTest;
-import io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SecurityProtocol;
 import io.confluent.connect.elasticsearch.helper.ElasticsearchContainer;
 
 import org.apache.kafka.common.config.SslConfigs;
@@ -28,14 +27,12 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-import java.util.Collections;
 import java.util.List;
 import java.util.Map;
 
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_PASSWORD_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_URL_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.CONNECTION_USERNAME_CONFIG;
-import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SECURITY_PROTOCOL_CONFIG;
 import static io.confluent.connect.elasticsearch.ElasticsearchSinkConnectorConfig.SSL_CONFIG_PREFIX;
 
 
@@ -58,20 +55,27 @@ public static void setupBeforeAll() {
    */
   @Test
   public void testSecureConnectionVerifiedHostname() throws Throwable {
-    // Use IP address here because that's what the certificates allow
-    String address = container.getConnectionUrl();
-    address = address.replace(container.getContainerIpAddress(), container.hostMachineIpAddress());
+    // Use container IP address here because that's what the certificates allow
+    String address = container.getConnectionUrl(false);
     log.info("Creating connector for {}.", address);
 
     props.put(CONNECTION_URL_CONFIG, address);
-    addSslProps();
+    container.addSslProps(props);
 
     helperClient = container.getHelperClient(props);
 
     // Start connector
     runSimpleTest(props);
   }
 
+  @Override
+  protected Map<String, String> createProps() {
+    props = super.createProps();
+    props.put(CONNECTION_USERNAME_CONFIG, ELASTIC_MINIMAL_PRIVILEGES_NAME);
+    props.put(CONNECTION_PASSWORD_CONFIG, ELASTIC_MINIMAL_PRIVILEGES_PASSWORD);
+    return props;
+  }
+
   @Test
   public void testSecureConnectionHostnameVerificationDisabled() throws Throwable {
     // Use 'localhost' here that is not in self-signed cert
@@ -80,7 +84,7 @@ public void testSecureConnectionHostnameVerificationDisabled() throws Throwable
     log.info("Creating connector for {}", address);
 
     props.put(CONNECTION_URL_CONFIG, address);
-    addSslProps();
+    container.addSslProps(props);
 
     // disable hostname verification
     props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_ENDPOINT_IDENTIFICATION_ALGORITHM_CONFIG, "");
@@ -90,15 +94,4 @@ public void testSecureConnectionHostnameVerificationDisabled() throws Throwable
     // Start connector
     runSimpleTest(props);
   }
-
-  private void addSslProps() {
-    props.put(SECURITY_PROTOCOL_CONFIG, SecurityProtocol.SSL.name());
-    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_KEYSTORE_LOCATION_CONFIG, container.getKeystorePath());
-    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_KEYSTORE_PASSWORD_CONFIG, container.getKeystorePassword());
-    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_TRUSTSTORE_LOCATION_CONFIG, container.getTruststorePath());
-    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_TRUSTSTORE_PASSWORD_CONFIG, container.getTruststorePassword());
-    props.put(SSL_CONFIG_PREFIX + SslConfigs.SSL_KEY_PASSWORD_CONFIG, container.getKeyPassword());
-    props.put(CONNECTION_USERNAME_CONFIG, ELASTIC_MINIMAL_PRIVILEGES_NAME);
-    props.put(CONNECTION_PASSWORD_CONFIG, ELASTIC_MINIMAL_PRIVILEGES_PASSWORD);
-  }
 }