Commit 64bb4402 authored by Robert Yokota's avatar Robert Yokota
Browse files

CC-1385: Make Elasticsearch connector version-aware; enhance connector to use text type with ES 5+ 

The ES connector now queries for the version.  The version is stored in
an instance of ElasticsearchClient, which is passed around so that
it can be queried at the appropriate times (such as when inferring
schema mappings).

As part of this change, the old Jest client is now wrapped by a
higher-level client object.  Also, all the Jest dependencies have been
isolated to the io.confluent.connect.elasticsearch.jest package.  This
makes the third-party dependencies more clear and will facilitate moving
to a different ES client library in the future should we choose to do
so.

These changes have been tested with ES 2.x, 5.x, and 6.x.
parent e3049c93

checkstyle/suppressions.xml

+1 −1
Original line number Diff line number Diff line
@@ -15,7 +15,7 @@ 
    <!-- TODO: Undecided if this is too much -->

    <suppress

            checks="(ClassDataAbstractionCoupling)"

            files="(BulkProcessor).java"

            files="(BulkProcessor|JestElasticsearchClient).java"

    />



    <!-- TODO: Pass some parameters in common config object? -->

pom.xml

+36 −10
Original line number Diff line number Diff line
@@ -39,11 +39,16 @@ 
    </scm>



    <properties>

        <es.version>6.0.0</es.version>

        <lucene.version>7.0.1</lucene.version>

        <!--

        <es.version>5.0.0</es.version>

        <lucene.version>6.2.0</lucene.version>

        <es.version>2.4.1</es.version>

        <lucene.version>5.5.2</lucene.version>

        <jna.version>4.2.1</jna.version>

        <hamcrest.version>2.0.0.0</hamcrest.version>

        <jest.version>2.0.0</jest.version>

        -->

        <hamcrest.version>1.3</hamcrest.version>

        <jest.version>5.3.3</jest.version>

        <confluent.maven.repo>http://packages.confluent.io/maven/</confluent.maven.repo>

    </properties>
@@ -71,15 +76,9 @@ 
            <artifactId>jest</artifactId>

            <version>${jest.version}</version>

        </dependency>

        <dependency>

            <groupId>net.java.dev.jna</groupId>

            <artifactId>jna</artifactId>

            <version>${jna.version}</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.hamcrest</groupId>

            <artifactId>hamcrest-junit</artifactId>

            <artifactId>hamcrest-all</artifactId>

            <version>${hamcrest.version}</version>

            <scope>test</scope>

        </dependency>
@@ -101,6 +100,20 @@ 
            <version>${lucene.version}</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.apache.logging.log4j</groupId>

            <artifactId>log4j-api</artifactId>

            <version>2.7</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.apache.logging.log4j</groupId>

            <artifactId>log4j-core</artifactId>

            <version>2.7</version>

            <scope>test</scope>

        </dependency>

        <!-- For ES 2.x -->

        <!--

        <dependency>

            <groupId>org.elasticsearch</groupId>

            <artifactId>elasticsearch</artifactId>
@@ -108,12 +121,25 @@ 
            <scope>test</scope>

            <type>test-jar</type>

        </dependency>

        -->

        <dependency>

            <groupId>org.elasticsearch.test</groupId>

            <artifactId>framework</artifactId>

            <version>${es.version}</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.elasticsearch</groupId>

            <artifactId>elasticsearch</artifactId>

            <version>${es.version}</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.elasticsearch.plugin</groupId>

            <artifactId>transport-netty4-client</artifactId>

            <version>${es.version}</version>

            <scope>test</scope>

        </dependency>

        <dependency>

            <groupId>org.apache.lucene</groupId>

            <artifactId>lucene-expressions</artifactId>

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

+10 −63
Original line number Diff line number Diff line
@@ -16,82 +16,29 @@ 


package io.confluent.connect.elasticsearch;



import com.fasterxml.jackson.databind.ObjectMapper;

import com.fasterxml.jackson.databind.node.ObjectNode;



import org.slf4j.Logger;

import org.slf4j.LoggerFactory;



import java.io.IOException;

import java.util.ArrayList;

import java.util.List;



import io.confluent.connect.elasticsearch.bulk.BulkClient;

import io.confluent.connect.elasticsearch.bulk.BulkRequest;

import io.confluent.connect.elasticsearch.bulk.BulkResponse;

import io.searchbox.client.JestClient;

import io.searchbox.core.Bulk;

import io.searchbox.core.BulkResult;



public class BulkIndexingClient implements BulkClient<IndexableRecord, Bulk> {



  private static final Logger LOG = LoggerFactory.getLogger(BulkIndexingClient.class);

import java.io.IOException;

import java.util.List;



  private static final ObjectMapper OBJECT_MAPPER = new ObjectMapper();

public class BulkIndexingClient implements BulkClient<IndexableRecord, BulkRequest> {



  private final JestClient client;

  private final ElasticsearchClient client;



  public BulkIndexingClient(JestClient client) {

  public BulkIndexingClient(ElasticsearchClient client) {

    this.client = client;

  }



  @Override

  public Bulk bulkRequest(List<IndexableRecord> batch) {

    final Bulk.Builder builder = new Bulk.Builder();

    for (IndexableRecord record : batch) {

      builder.addAction(record.toBulkableAction());

    }

    return builder.build();

  public BulkRequest bulkRequest(List<IndexableRecord> batch) {

    return client.createBulkRequest(batch);

  }



  @Override

  public BulkResponse execute(Bulk bulk) throws IOException {

    final BulkResult result = client.execute(bulk);



    if (result.isSucceeded()) {

      return BulkResponse.success();

    }



    boolean retriable = true;



    final List<Key> versionConflicts = new ArrayList<>();

    final List<String> errors = new ArrayList<>();



    for (BulkResult.BulkResultItem item : result.getItems()) {

      if (item.error != null) {

        final ObjectNode parsedError = (ObjectNode) OBJECT_MAPPER.readTree(item.error);

        final String errorType = parsedError.get("type").asText("");

        if ("version_conflict_engine_exception".equals(errorType)) {

          versionConflicts.add(new Key(item.index, item.type, item.id));

        } else if ("mapper_parse_exception".equals(errorType)) {

          retriable = false;

          errors.add(item.error);

        } else {

          errors.add(item.error);

        }

      }

    }



    if (!versionConflicts.isEmpty()) {

      LOG.debug("Ignoring version conflicts for items: {}", versionConflicts);

      if (errors.isEmpty()) {

        // The only errors were version conflicts

        return BulkResponse.success();

      }

    }



    final String errorInfo = errors.isEmpty() ? result.getErrorMessage() : errors.toString();



    return BulkResponse.failure(retriable, errorInfo);

  public BulkResponse execute(BulkRequest bulk) throws IOException {

    return client.executeBulk(bulk);

  }



}

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

0 → 100644
+97 −0
Original line number Diff line number Diff line 
/**

 * Copyright 2018 Confluent Inc.

 *

 * Licensed under the Apache License, Version 2.0 (the "License"); you may not

 * use this file except in compliance with the License. You may obtain a copy of

 * the License at

 *

 * http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT

 * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the

 * License for the specific language governing permissions and limitations under

 * the License.

 **/



package io.confluent.connect.elasticsearch;



import com.google.gson.JsonObject;

import io.confluent.connect.elasticsearch.bulk.BulkRequest;

import io.confluent.connect.elasticsearch.bulk.BulkResponse;

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



import java.io.IOException;

import java.util.List;

import java.util.Set;



public interface ElasticsearchClient {



  enum Version {

    ONE, TWO, FIVE, SIX

  }



  /**

   * Gets the Elasticsearch version.

   *

   * @return the version

   */

  Version getVersion();



  /**

   * Creates indices.

   *

   * @param indices the set of index names to create

   */

  void createIndices(Set<String> indices);



  /**

   * Creates an explicit mapping.

   *

   * @param index the index to write

   * @param type the type for which to create the mapping

   * @param schema the schema used to infer the mapping

   * @throws IOException from underlying client

   */

  void createMapping(String index, String type, Schema schema) throws IOException;



  /**

   * Gets the JSON mapping for the given index and type. Returns {@code null} if it does not exist.

   *

   * @param index the index

   * @param type the type

   * @throws IOException from underlying client

   */

  JsonObject getMapping(String index, String type) throws IOException;



  /**

   * Creates a bulk request for the list of {@link IndexableRecord} records.

   *

   * @param batch the list of records

   * @return the bulk request

   */

  BulkRequest createBulkRequest(List<IndexableRecord> batch);



  /**

   * Executes a bulk action.

   *

   * @param bulk the bulk request

   * @return the bulk response

   */

  BulkResponse executeBulk(BulkRequest bulk) throws IOException;



  /**

   * Executes a search.

   *

   * @param query the search query

   * @param index the index to search

   * @param type the type to search

   * @return the search result

   */

  JsonObject search(String query, String index, String type) throws IOException;



  /**

   * Shuts down the client.

   */

  void shutdown();

}

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

+1 −19
Original line number Diff line number Diff line
@@ -16,11 +16,6 @@ 


package io.confluent.connect.elasticsearch;



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



import java.util.HashMap;

import java.util.Map;



public class ElasticsearchSinkConnectorConstants {

  public static final String MAP_KEY = "key";

  public static final String MAP_VALUE = "value";
@@ -34,19 +29,6 @@ public class ElasticsearchSinkConnectorConstants { 
  public static final String FLOAT_TYPE = "float";

  public static final String DOUBLE_TYPE = "double";

  public static final String STRING_TYPE = "string";

  public static final String TEXT_TYPE = "text";

  public static final String DATE_TYPE = "date";



  static final Map<Type, String> TYPES = new HashMap<>();



  static {

    TYPES.put(Type.BOOLEAN, BOOLEAN_TYPE);

    TYPES.put(Type.INT8, BYTE_TYPE);

    TYPES.put(Type.INT16, SHORT_TYPE);

    TYPES.put(Type.INT32, INTEGER_TYPE);

    TYPES.put(Type.INT64, LONG_TYPE);

    TYPES.put(Type.FLOAT32, FLOAT_TYPE);

    TYPES.put(Type.FLOAT64, DOUBLE_TYPE);

    TYPES.put(Type.STRING, STRING_TYPE);

    TYPES.put(Type.BYTES, BINARY_TYPE);

  }

}