CC-12445: Replace the JestClient with the ElasticsearchRestClient #468

levzem · 2020-11-10T01:14:34Z

Signed-off-by: Lev Zemlyanov lev@confluent.io

Problem

the jest client is outdated and needs to be replaced by the supported and maintained ES high level rest client

Solution

replace the client

major points

BulkProcessor is no longer required as ES provides its own
all Jest classes are no longer necessary
ElasticsearchWriter was absorbed into SinkTask
retries are supported for all operations
some configs are deprecated/not used anymore WIP, full list TBD)
- type.name as it is deprecated as of ES 7.x

Does this solution apply anywhere else?

yes
no

If yes, where?

Test Strategy

an additional manual upgrade test was performed. write records to ES cloud with a v10.0.2, then upgrade to the new client and try writing to the same ES cloud index

Testing done:

Unit tests
Integration tests
System tests
Manual tests

Release Plan

master because this is a major change

kkonstantine

Thanks @levzem
This is going to be a tremendous upgrade!

Took a first quick pass over half the files and left a few comments. Will have to return soon for a more careful read.

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

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

pom.xml

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

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

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

avocader · 2020-11-18T09:59:44Z

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

+          url ->
+              credentialsProvider.setCredentials(
+                  new AuthScope(new HttpHost(url)),
+                  new UsernamePasswordCredentials(config.username(), config.password().value())


Can password be null?

nope :), https://github.com/confluentinc/kafka-connect-elasticsearch/blob/master/src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkConnectorConfig.java#L657-L659

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

avocader · 2020-11-18T10:07:40Z

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

+          try {
+            client.indices().create(request, RequestOptions.DEFAULT);
+          } catch (ElasticsearchStatusException | IOException e) {
+            if (!e.getMessage().contains(RESOURCE_ALREADY_EXISTS_EXCEPTION)) {


Is it still possible to get this exception if we already checked whether the index exists earlier?

we have seen it before in escalations and this maintains the previous behavior of the JestClient so I kept it just to be safe

unfortunately we have not been able to understand why this happens

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

avocader · 2020-11-18T10:15:40Z

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

+   * @param id the document id
+   * @param record the record
+   */
+  private void addToRecordMap(String id, SinkRecord record) {


I see that it's called once, do we need this method?

leftover from BulkProcessor implementation, while yes it is only called once, given it is wrapped in a null check it is probably better to abstract it away in case of future use

avocader · 2020-11-18T10:17:05Z

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

+
+      @Override
+      public void afterBulk(long executionId, BulkRequest request, Throwable failure) {
+        error.compareAndSet(null, new ConnectException("Bulk request failed.", failure));


Are we keeping the document in the recordMap?

this will fail the task so it wont matter if we empty the map or not, but I added it regardless

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

avocader · 2020-11-18T10:35:03Z

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

+            response.getIndex()
+        );
+
+        reportBadRecord(response);


The method's Javadoc and the warn message says we are ignoring version conflict,s but it seems like we are still reporting this record. Is that intentional?

reporting the record is the same as ignoring it (reporting a bad record does not fail anything), but I figured the user should be know which records had version conflicts in case they want to remediate manually somehow

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

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

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

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

avocader · 2020-12-02T01:52:37Z

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

+    return Version.getVersion();
+  }
+
+  private void checkMapping(SinkRecord record) {


This method name is a bit misleading IMO, maybe consider renaming it to maybeCreateMapping?

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

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

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

avocader · 2020-12-02T02:10:37Z

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

+  }
+
+  @Test
+  public void testCallWithRetriesSomeRetries() throws Exception {


nit: add a unit-test that ensures wait time is always lesser than MAX_RETRY_TIME_MS?

already tested by https://github.com/confluentinc/kafka-connect-elasticsearch/blob/master/src/test/java/io/confluent/connect/elasticsearch/RetryUtilTest.java#L57
and
https://github.com/confluentinc/kafka-connect-elasticsearch/blob/master/src/test/java/io/confluent/connect/elasticsearch/RetryUtilTest.java#L23-L46

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

avocader

I left a few non-critical suggestions, LGTM otherwise, thanks @levzem !

kkonstantine

Another round of comments. Mostly done. I have a few more files to check. Mostly tests.

pom.xml

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

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

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

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

levzem · 2020-12-02T19:10:37Z

@kkonstantine, addressed all review comments, ready for another pass!

Signed-off-by: Lev Zemlyanov <lev@confluent.io>

kkonstantine

Very nice work.
Left a few more minor comments.
Also please make sure to run mvn javadoc:javadoc and fix the javadoc issues because they'll break during the release.

But this LGTM! 👏

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

kkonstantine · 2020-12-03T02:19:30Z

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

+    client.close();
+  }
+
+  @Test(expected = ConnectException.class)


It'd be nice to start using assertThrows and available hamcrest matchers for maps etc. But we can consider on a follow up.

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

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

Signed-off-by: Lev Zemlyanov <lev@confluent.io>

levzem requested a review from a team as a code owner November 10, 2020 01:14

This was referenced Nov 11, 2020

CC-12597: remove topic.index.map #470

Merged

HCD-448: Adding support for size-based batching of sink records. #442

Closed

levzem force-pushed the client branch 5 times, most recently from 10f2413 to a57b0da Compare November 17, 2020 01:03

kkonstantine reviewed Nov 18, 2020

View reviewed changes