Commit b3d027b4 authored by Randall Hauch's avatar Randall Hauch
Browse files

MINOR: Use correct config type for recently added password field

parent 23a31fc9

docs/configuration_options.rst

+5 −3
Original line number Diff line number Diff line
@@ -14,15 +14,17 @@ Connector 
  * Importance: high



``connection.username``

  The username used to authenticate with Elasticsearch. The default is null, and authentication will only be performed if both the username and password are non-null.

  The username used to authenticate with Elasticsearch. The default is the null, and authentication will only be performed if  both the username and password are non-null.



  * Type: string

  * Default: null

  * Importance: medium



``connection.password``

  The password used to authenticate with Elasticsearch. The default is null, and authentication will only be performed if both the username and password are non-null.

  The password used to authenticate with Elasticsearch. The default is the null, and authentication will only be performed if  both the username and password are non-null.



  * Type: string

  * Type: password

  * Default: null

  * Importance: medium



``batch.size``

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java

+1 −1
Original line number Diff line number Diff line
@@ -180,7 +180,7 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig { 
        "Connection Username"

    ).define(

        CONNECTION_PASSWORD_CONFIG,

        Type.STRING,

        Type.PASSWORD,

        null,

        Importance.MEDIUM,

        CONNECTION_PASSWORD_DOC,

src/main/java/io/confluent/connect/elasticsearch/jest/JestElasticsearchClient.java

+3 −2
Original line number Diff line number Diff line
@@ -46,6 +46,7 @@ import io.searchbox.indices.mapping.GetMapping; 
import io.searchbox.indices.mapping.PutMapping;

import org.apache.http.HttpHost;

import org.apache.kafka.common.config.ConfigException;

import org.apache.kafka.common.config.types.Password;

import org.apache.kafka.connect.data.Schema;

import org.apache.kafka.connect.errors.ConnectException;

import org.slf4j.Logger;
@@ -124,7 +125,7 @@ public class JestElasticsearchClient implements ElasticsearchClient { 


      final String username = config.getString(

          ElasticsearchSinkConnectorConfig.CONNECTION_USERNAME_CONFIG);

      final String password = config.getString(

      final Password password = config.getPassword(

          ElasticsearchSinkConnectorConfig.CONNECTION_PASSWORD_CONFIG);



      List<String> address =
@@ -134,7 +135,7 @@ public class JestElasticsearchClient implements ElasticsearchClient { 
          .readTimeout(readTimeout)

          .multiThreaded(true);

      if (username != null && password != null) {

        builder.defaultCredentials(username, password)

        builder.defaultCredentials(username, password.value())

            .preemptiveAuthTargetHosts(address.stream()

                .map(addr -> HttpHost.create(addr)).collect(Collectors.toSet()));

      }