[Gobblin-17] Add Elasticsearch writer (rest + transport) #2419

shirshanka · 2018-08-07T08:18:57Z

Dear Gobblin maintainers,

Please accept this PR. I understand that it will not be reviewed until I have checked off all the steps below!

JIRA

My PR addresses the following Gobblin JIRA issues and references them in the PR title. For example, "[GOBBLIN-XXX] My Gobblin PR"
- https://issues.apache.org/jira/browse/GOBBLIN-17

Description

Here are some details about my PR, including screenshots (if applicable):
Adds two elasticsearch writer types (transport and rest)

Tests

My PR adds the following unit tests OR does not need testing for this extremely good reason:
ElasticsearchRestClientWriterTest
ElasticsearchTransportClientWriterTest

Commits

My commits all reference JIRA issues in their subject lines, and I have squashed multiple commits if they address the same issue. In addition, my commits follow the guidelines from "How to write a good git commit message":
1. Subject is separated from body by a blank line
2. Subject is limited to 50 characters
3. Subject does not end with a period
4. Subject uses the imperative mood ("add", not "adding")
5. Body wraps at 72 characters
6. Body explains "what" and "why", not "how"

zxcware · 2018-08-07T23:50:24Z

gradle/scripts/dependencyDefinitions.gradle

@@ -47,7 +47,7 @@ ext.externalDependency = [
    "datanucleusCore": "org.datanucleus:datanucleus-core:3.2.10",
    "datanucleusRdbms": "org.datanucleus:datanucleus-rdbms:3.2.9",
    "eventhub": "com.microsoft.azure:azure-eventhubs:0.9.0",
-    "guava": "com.google.guava:guava:15.0",
+    "guava": "com.google.guava:guava:18.0",


It's not a good time to upgrade guava to version 18, as the Hadoop API depends on guava APIs that are removed since guava 16.

Elasticsearch client breaks without it :(

@zxcware do you know which Hadoop API breaks guava version upgrade?

yukuai518 · 2018-08-08T06:29:46Z

gradle/scripts/dependencyDefinitions.gradle

@@ -47,7 +47,7 @@ ext.externalDependency = [
    "datanucleusCore": "org.datanucleus:datanucleus-core:3.2.10",
    "datanucleusRdbms": "org.datanucleus:datanucleus-rdbms:3.2.9",
    "eventhub": "com.microsoft.azure:azure-eventhubs:0.9.0",
-    "guava": "com.google.guava:guava:15.0",
+    "guava": "com.google.guava:guava:18.0",


@zxcware do you know which Hadoop API breaks guava version upgrade?

yukuai518 · 2018-08-08T06:36:35Z

...elasticsearch/src/main/java/org/apache/gobblin/elasticsearch/writer/ResponseErrorLogger.java

+/**
+ * An interface to log errors during responses
+ */
+public interface ResponseErrorLogger {


Maybe rename logError to logException? And rename class to ResponseExceptionLogger?

yukuai518 · 2018-08-08T06:39:29Z

...in-elasticsearch/src/test/java/org/apache/gobblin/elasticsearch/ElasticsearchTestServer.java

+      int port = testServer.getPort();
+//      int serverPort = couchbaseTestServer.getServerPort();
+
+      /**


why we need to keep this block commented out?

yukuai518 · 2018-08-08T06:40:01Z

...in-elasticsearch/src/test/java/org/apache/gobblin/elasticsearch/ElasticsearchTestServer.java

+
+  private void fillServerPort()
+  {
+    /**


why this block needs to be commented out?

yukuai518 · 2018-08-08T07:02:56Z

...lasticsearch/src/main/java/org/apache/gobblin/elasticsearch/writer/FutureCallbackHolder.java

+          throws InterruptedException, ExecutionException, TimeoutException {
+        Pair<WriteResponse, Throwable> writeResponseThrowablePair = writeResponseQueue.poll(timeout, unit);
+        if (writeResponseThrowablePair == null) {
+          throw new TimeoutException("Timeout exceeded while waiting for future to be done");


In this timeout case, do we need to invoke the callback?

This is a client specifying how long they want to wait on the future... they could decide to issue this call again... so IMO the callback should only be fired when the underlying call to the destination succeeds or fails.

abti

Mostly minor comments. Also, please include your shading work for guava.

abti · 2018-08-09T21:56:30Z

gobblin-core-base/src/main/java/org/apache/gobblin/writer/BytesBoundedBatch.java

@@ -62,7 +62,11 @@ void append (D record) {
      records.add(record);
    }

-    boolean hasRoom (D record) {
+    boolean hasRoom (D record, LargeMessagePolicy largeMessagePolicy) {
+      if (records.isEmpty() && largeMessagePolicy == LargeMessagePolicy.ATTEMPT) {


throw exception on LargeMessagePolicy.FAIL per contract?

done now... but in a different place (tryAppend), not hasRoom.

abti · 2018-08-09T21:59:44Z

gobblin-modules/gobblin-elasticsearch/build.gradle

+  compile project(":gobblin-utility")
+  compile project(":gobblin-metrics-libs:gobblin-metrics")
+
+  compile "org.elasticsearch.client:transport:5.6.8"


why is it not part of externalDependency spec and referred here as: externalDependency.elasticSearchClient etc

abti · 2018-08-09T22:01:14Z

gobblin-modules/gobblin-elasticsearch/scripts/install_test_deps.sh

+#
+
+TARGET_DIR="test-elasticsearch"
+ES_VERSION=5.6.8


should this be passed along from gradle script as param, where gralde script 'knows' about it through project properties (as versions for deps are defined in gradle scripts)

Not sure if it can be made that generic and configurable... the download locations for different versions of elasticsearch were different when I spot-checked a couple.

Not going to do this... too much work.

nw, create a Jira :)

abti · 2018-08-09T22:04:36Z

.../src/main/java/org/apache/gobblin/elasticsearch/typemapping/AvroGenericRecordTypeMapper.java

+    }
+    catch (Exception e) {
+      //TODO
+      throw new NoSuchFieldException("Could not find field ");


propagate e with NoSuchFieldException?

abti · 2018-08-09T22:08:17Z

...elasticsearch/src/main/java/org/apache/gobblin/elasticsearch/typemapping/JsonSerializer.java

+import com.typesafe.config.Config;
+
+
+public interface JsonSerializer<T> extends Closeable {


why is this specific to ElasticSearch? looks like helpful common util?

I think our ser-de + type-mapping approach needs a re-think. Didn't want to take on that effort while writing this integration. We can use this approach as one strawman when we design the framework wide support for multiple formats.

abti · 2018-08-09T22:14:26Z

...ticsearch/src/main/java/org/apache/gobblin/elasticsearch/writer/ElasticsearchRestWriter.java

+        String keyStoreType = ConfigUtils
+            .getString(config, ElasticsearchWriterConfigurationKeys.ELASTICSEARCH_WRITER_SSL_KEYSTORE_TYPE,
+                ElasticsearchWriterConfigurationKeys.ELASTICSEARCH_WRITER_SSL_KEYSTORE_TYPE_DEFAULT);
+        String keyStoreFilePassword = ConfigUtils


Use password manager? Its a one line change but makes it secure. Or create a ticket and assign to me, I will do it.

abti · 2018-08-09T22:30:25Z

...in-elasticsearch/src/test/java/org/apache/gobblin/elasticsearch/ElasticsearchTestServer.java

+
+
+  private Process elasticProcess;
+  private int _port = 9200;


how about using a free port available at the time, like: https://stackoverflow.com/questions/51099027/find-free-port-in-java
we had frequent transient test failures in service / cluster test cases when they were using fixed ports to setup pre-requisite infra to run tests

abti

+1

abti · 2018-08-31T22:20:14Z

gobblin-core-base/src/main/java/org/apache/gobblin/writer/SequentialBasedBatchAccumulator.java

@@ -240,12 +261,16 @@ public void close() {
  public void flush() {
    try {
      ArrayList<Batch> batches = this.incomplete.all();
-      LOG.info ("flush on {} batches", batches.size());
+      int numOutstandingRecords = 0;


abti · 2018-08-31T22:22:03Z

gobblin-modules/gobblin-elasticsearch/build.gradle

+
+task installTestDependencies(type:Exec) {
+  workingDir "${project.rootDir}/gobblin-modules/gobblin-elasticsearch/"
+  commandLine './scripts/install_test_deps.sh'


nit: fail build for non-zero exit?

abti · 2018-08-31T22:26:22Z

gobblin-modules/gobblin-elasticsearch/scripts/install_test_deps.sh

+#
+
+TARGET_DIR="test-elasticsearch"
+ES_VERSION=5.6.8


nw, create a Jira :)

abti · 2018-08-31T22:33:29Z

...ticsearch/src/main/java/org/apache/gobblin/elasticsearch/writer/ElasticsearchWriterBase.java

+      this.hostAddresses = new ArrayList<>(hosts.size());
+      for (String host : hosts) {
+
+        List<String> hostSplit = hostSplitter.splitToList(host);


There should be some Apache util to do this :)

thadguidry · 2021-08-04T14:48:56Z

I don't see this writer mentioned in the official docs...is there a Jira issue already logged for it? I'm thinking it might be good to at least document this lightly so folks know it's available now in releases?

shirshanka force-pushed the elastic branch 2 times, most recently from 39adda3 to 2d2a3ce Compare August 7, 2018 22:43

zxcware suggested changes Aug 7, 2018

View reviewed changes

yukuai518 reviewed Aug 8, 2018

View reviewed changes

abti reviewed Aug 9, 2018

View reviewed changes

shirshanka force-pushed the elastic branch 3 times, most recently from 108c3dd to f25f95c Compare August 21, 2018 19:07

[Gobblin-17] Add Elasticsearch writer (rest + transport)

1182be1

shirshanka force-pushed the elastic branch from 7aa2f3d to 1182be1 Compare August 31, 2018 01:48

abti approved these changes Aug 31, 2018

View reviewed changes

asfgit closed this in f1bc746 Sep 1, 2018

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[Gobblin-17] Add Elasticsearch writer (rest + transport) #2419

[Gobblin-17] Add Elasticsearch writer (rest + transport) #2419

shirshanka commented Aug 7, 2018

zxcware Aug 7, 2018

shirshanka Aug 8, 2018

yukuai518 Aug 8, 2018

yukuai518 Aug 8, 2018

yukuai518 Aug 8, 2018

yukuai518 Aug 8, 2018 •

edited

yukuai518 Aug 8, 2018

yukuai518 Aug 8, 2018

shirshanka Aug 15, 2018

abti left a comment

abti Aug 9, 2018

shirshanka Aug 21, 2018

abti Aug 9, 2018

abti Aug 9, 2018

shirshanka Aug 21, 2018

shirshanka Aug 31, 2018

abti Aug 31, 2018

abti Aug 9, 2018

abti Aug 9, 2018

shirshanka Aug 21, 2018

abti Aug 9, 2018

abti Aug 9, 2018

abti left a comment

abti Aug 31, 2018

abti Aug 31, 2018

abti Aug 31, 2018

abti Aug 31, 2018

thadguidry commented Aug 4, 2021

		import com.typesafe.config.Config;


		public interface JsonSerializer<T> extends Closeable {

[Gobblin-17] Add Elasticsearch writer (rest + transport) #2419

[Gobblin-17] Add Elasticsearch writer (rest + transport) #2419

Conversation

shirshanka commented Aug 7, 2018

JIRA

Description

Tests

Commits

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

yukuai518 Aug 8, 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

abti 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

abti 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

thadguidry commented Aug 4, 2021

yukuai518 Aug 8, 2018 •

edited