CC-1385: Enhance connector to use text type with ES 5+ #169

rayokota · 2018-01-29T19:29:25Z

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.

…r 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.

rhauch

Initial review with some suggestions/comments/questions. I'll want to do another round after those are addressed and we sort out the build issue.

rhauch · 2018-01-29T20:22:15Z

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

+  private final Version version;
+
+  @VisibleForTesting
+  public JestElasticsearchClient(String address) {


Can this be package-level scope?

I tend to prefer protected unless I have a good reason to leave package-private. Several refactorings in hdfs and storage-common showed that something might need to be used that way for testing but not reside exactly in the same package (for various reasons). Not feeling too strong about this though.

Currently it needs to be public as it is used in a different package.

rhauch · 2018-01-29T20:22:47Z

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

        return inferPrimitive(esType, schema.defaultValue());
    }
  }

+  @VisibleForTesting
+  public static String getElasticsearchType(ElasticsearchClient client, Schema.Type schemaType) {


Can this be package-level scope?

Yes, I'll change it

rhauch · 2018-01-29T20:25:33Z

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

+
+  private Version getServerVersion() throws IOException {
+    // Default to newest version for forward compatibility
+    Version defaultVersion = Version.SIX;


IIUC, ES 6 removes the old string type that was deprecated in 5. If that's true, then how is defaulting to 6 forward compatible?

It's forward compatible for ES 7.x+, meaning that the ES 6 functionality will be retained (using text instead of string, which only works for ES 2 and 5).

rhauch · 2018-01-29T20:29:57Z

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

+      return defaultVersion;
+    }
+
+    String esVersion = nodeRoot.get("version").getAsString();


Does this should work for all of the ES versions and the above logic is really handling exceptional cases? If so, then perhaps this method should have doc that explains that. And, the log messages should maybe be a bit more descriptive so that we can more easily tell which path the logic takes.

Yes, it should work for all versions. I'll make the errors more descriptive.

rhauch · 2018-01-29T20:33:07Z

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

+  /**
+   * Gets the Elasticsearch version.
+   *
+   * @return the version


Never null?

Yes, I'll add a comment

rhauch · 2018-01-29T20:34:56Z

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

+   * @param query the search query
+   * @param index the index to search
+   * @param type the type to search
+   * @return the search result


Missing @throws IOException if the client cannot execute the request

I'll add it.

rhauch · 2018-01-29T20:35:57Z

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

+  /**
+   * Shuts down the client.
+   */
+  void shutdown();


Have you considered extending AutoCloseable and then overriding the close() method here to throw no exceptions?

Good idea, I'll do this.

rhauch · 2018-01-29T20:43:50Z

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

+import java.util.Map;
+import java.util.Set;
+
+public class JestElasticsearchClient implements ElasticsearchClient {


Can we add unit tests for this? Might have to mock the JestClient, in which case we'd want a constructor that takes the JestClient so we can pass in the mock. This class' logic is not trivial, and it seems like it'd be better to test this logic in isolation. Thoughts?

I'll look into adding some tests for this.

rhauch · 2018-01-29T20:46:21Z

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

+          if (!result.isSucceeded()) {
+            String msg = result.getErrorMessage() != null ? ": " + result.getErrorMessage() : "";
+            throw new ConnectException("Could not create index '" + index + "'" + msg);
+          }


Is it possible that two tasks attempt to look for the same index, neither finds it, and so both attempt to create the same index? If so, would it make sense if the create request fails to first check and see if it exists (perhaps again) and to fail only if the index does not already exist?

This is the current behavior, but yes, I can add an additional check before throwing the exception.

rhauch · 2018-01-29T20:49:16Z

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

+    try {
+      client.close();
+    } catch (IOException e) {
+      LOG.warn("Could not close client");


How about including at least the exception message in the log message?

kkonstantine

Looks good! Thanks @rayokota

Gave it a first pass, noticing mainly packaging and stylistic things.

kkonstantine · 2018-01-29T22:35:06Z

pom.xml

@@ -39,11 +39,16 @@
    </scm>

    <properties>
+        <es.version>6.0.0</es.version>
+        <lucene.version>7.0.1</lucene.version>
+        <!--


nit: I assume commenting out this section is temporary?

kkonstantine · 2018-01-29T22:44:05Z

pom.xml

+            <scope>test</scope>
+        </dependency>
+        <!-- For ES 2.x -->
+        <!--


nit: don't forget to remove commented out section.

kkonstantine · 2018-01-29T22:54:08Z

pom.xml

@@ -101,19 +100,46 @@
            <version>${lucene.version}</version>
            <scope>test</scope>
        </dependency>
+        <dependency>
+            <groupId>org.apache.logging.log4j</groupId>
+            <artifactId>log4j-api</artifactId>


Normally we use slf4j-log4j12 (which also carries slf4j-api) and they are available by inheriting from the common repo.

This is test scope and was necessary for working with the new ES test libs. However, now that I've downgraded the ES test lib I will remove this.

kkonstantine · 2018-01-29T23:01:46Z

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

    try {
      log.info("Starting ElasticsearchSinkTask.");

      ElasticsearchSinkConnectorConfig config = new ElasticsearchSinkConnectorConfig(props);
-      String type = config.getString(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG);
-      boolean ignoreKey =
+      final String type = config.getString(ElasticsearchSinkConnectorConfig.TYPE_NAME_CONFIG);


Were all these finals required by checkstyle? Otherwise it's a lot of noise that doesn't correspond to actual changes.

Additionally to that, the connectors style does not require to use final keyword for local variables.

These finals were recommended by checkstyle at one point but I think they are not needed any longer so I can remove.

kkonstantine · 2018-01-29T23:03:41Z

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

@@ -56,43 +56,27 @@
   * @param schema The schema used to infer mapping.
   * @throws IOException from underlying JestClient
   */
-  public static void createMapping(JestClient client, String index, String type, Schema schema)
+  public static void createMapping(ElasticsearchClient client,
+                                   String index,


indentation style is not the one we use in connectors.

Hmm, reformatting code with new coding style left this as is.

Frequently, to apply the actual style, you need to put it all in one line, select the whole line, and press Alt+Cmd+L

When I do that, it just leaves it all on one line. What formatting should it have?

In IntelliJ, checking the settings under: Preferences -> Code Style -> Java
tab Wrapping and Braces
boxes for Method alignment parameters and Method call arguments
and make sure New line after '(' is ticked, Place ')' on new line is ticked (when that's the case I believe Align when multiline does not matter, but better let it unticked too).

Got it, thanks

kkonstantine · 2018-01-29T23:26:35Z

src/main/java/io/confluent/connect/elasticsearch/bulk/BulkRequest.java

+
+package io.confluent.connect.elasticsearch.bulk;
+
+public interface BulkRequest {


How about a javadoc that this is a marker interface, and what is it for?

Yes, I will add JavaDoc for this.

kkonstantine · 2018-01-29T23:27:22Z

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

+
+public class JestBulkRequest implements BulkRequest {
+
+  private Bulk bulk;


Should this be final?

Yes, I will make final.

kkonstantine · 2018-01-29T23:29:35Z

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

+  private final Version version;
+
+  @VisibleForTesting
+  public JestElasticsearchClient(String address) {


I tend to prefer protected unless I have a good reason to leave package-private. Several refactorings in hdfs and storage-common showed that something might need to be used that way for testing but not reside exactly in the same package (for various reasons). Not feeling too strong about this though.

kkonstantine · 2018-01-29T23:31:32Z

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

+
+  private Version getServerVersion() throws IOException {
+    // Default to newest version for forward compatibility
+    Version defaultVersion = Version.SIX;


Another example where I'm not 100% immediately we are referring to ES version.

Yes, I will change to ES_V6.

kkonstantine · 2018-01-29T23:33:20Z

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

+    Version defaultVersion = Version.SIX;
+
+    NodesInfo info = new NodesInfo.Builder().addCleanApiParameter("version").build();
+    JsonObject result = this.client.execute(info).getJsonObject();


Again stylistic: Not sure if this was there already, but we tend to use this mainly within constructors.

I will change it

wicknicks

Thanks a lot for looking at this, @rayokota! Some minor comment.

wicknicks · 2018-01-30T22:52:03Z

pom.xml

+        <lucene.version>7.0.1</lucene.version>
+        <es.version>5.0.0</es.version>
+        <lucene.version>6.2.0</lucene.version>
+        -->


should we delete this section instead of leaving it commented.

I will remove.

wicknicks · 2018-01-30T22:53:26Z

pom.xml

        <dependency>
            <groupId>org.elasticsearch</groupId>
            <artifactId>elasticsearch</artifactId>
            <version>${es.version}</version>
            <scope>test</scope>
            <type>test-jar</type>
        </dependency>
+        <!-- For ES 5.x (requires Java 8) -->
+        <!--


I think we will have to change a bunch of things to move to Java 8, so it might be fine to delete the commented sections for now. And then tackle the Java 8 independently when the time comes.

wicknicks · 2018-01-30T22:54:51Z

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

+import java.util.List;
+import java.util.Set;
+
+public interface ElasticsearchClient extends AutoCloseable {


wicknicks · 2018-01-30T22:57:05Z

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

+  /**
+   * Shuts down the client.
+   */
+  void close();


Isn't close already defined in AutoCloseable?

I overrode it to not throw Exception

wicknicks · 2018-01-30T23:03:04Z

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

@@ -34,19 +29,6 @@
  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";


ES also has a keyword type. Should we support it?

I can add it now as a placeholder.

wicknicks · 2018-01-30T23:06:07Z

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

-        }
-      }
+    if (assignedTopics == null) {
+      throw new NullPointerException();


Can be replaced with a call to java.util.Objects#requireNonNull(obj)

Ok, I'll change it.

wicknicks · 2018-01-30T23:10:43Z

src/main/java/io/confluent/connect/elasticsearch/bulk/BulkRequest.java

+ * Implementations will typically hold state comprised of instances of classes that are
+ * specific to the client library.
+ */
+public interface BulkRequest {


Should we call this ConnectBulkRequest to make it easier to read whenever it is used in the same class with org.elasticsearch.action.bulk.BulkRequest?

I'm going to leave this for now as all the other classes in this package start with Bulk..(BulkClient, BulkProcessor, BulkResponse).

wicknicks · 2018-01-30T23:14:01Z

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

+  private final Version version;
+
+  // visible for testing
+  public JestElasticsearchClient(String address) {


should we have a @VisibleForTesting constructor which accepts a JestClient as an argument. That way the test can construct the JestClient and can be mocked appropriately.

Yes, I'll add this as part of some unit tests for the Jest client.

rayokota · 2018-01-31T21:55:29Z

I've added unit tests for the Jest client.

kkonstantine

Great job @rayokota ! My comments have been addressed. Given that others had lots of comments too, I'll defer the final decision to them.

rhauch

Very nice job, @rayokota. LGTM!

wicknicks

Thanks a lot, @rayokota! This looks great!

rayokota · 2018-02-08T19:20:21Z

Thanks for the review, @rhauch , @kkonstantine , and @wicknicks !

rayokota requested review from rhauch and wicknicks January 29, 2018 19:29

rhauch reviewed Jan 29, 2018

View reviewed changes

Downgrade Jest and ES test libs (for Java 7)

93de790

kkonstantine reviewed Jan 29, 2018

View reviewed changes

rayokota added 2 commits January 29, 2018 16:02

Reformat code with new coding style settings

e064acd

Incorporate feedback from code review

9446aa4

wicknicks reviewed Jan 30, 2018

View reviewed changes

Add unit tests for Jest client; more cleanup from code review

bc24949

joel-hamill mentioned this pull request Feb 1, 2018

DOCS-208 - Connect Elastic Quickstart not working #166

Merged

kkonstantine approved these changes Feb 7, 2018

View reviewed changes

rhauch approved these changes Feb 8, 2018

View reviewed changes

rhauch mentioned this pull request Feb 8, 2018

added a table of supported versions #173

Closed

wicknicks approved these changes Feb 8, 2018

View reviewed changes

rayokota merged commit 0ff9c21 into confluentinc:4.0.x Feb 8, 2018

This was referenced Feb 22, 2018

ES 5.0 compatibility #55

Closed

Support for Elasticsearch 6.0 #154

Closed

No handler for type [string] declared on field [after]"}] #160

Closed

rhauch mentioned this pull request May 18, 2018

Enable string fields for aggregation in Elasticsearch even if schema.ignore=false #211

Closed

levzem mentioned this pull request Nov 19, 2019

Added Jest result error message in case of create index exception #79

Closed


		package io.confluent.connect.elasticsearch.bulk;

		public interface BulkRequest {


		public class JestBulkRequest implements BulkRequest {

		private Bulk bulk;

CC-1385: Enhance connector to use text type with ES 5+ #169

CC-1385: Enhance connector to use text type with ES 5+ #169

Conversation

rayokota commented Jan 29, 2018

rhauch left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rayokota Jan 29, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kkonstantine left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

kkonstantine Jan 30, 2018 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

wicknicks left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

rayokota commented Jan 31, 2018

kkonstantine left a comment

Choose a reason for hiding this comment

rhauch left a comment

Choose a reason for hiding this comment

wicknicks left a comment

Choose a reason for hiding this comment

rayokota commented Feb 8, 2018

rayokota Jan 29, 2018 •

edited

kkonstantine Jan 30, 2018 •

edited