[FLINK-30998] Add optional exception handler to flink-connector-opensearch #11

lilyevsky · 2023-02-15T23:00:32Z

No description provided.

reta · 2023-02-15T23:46:41Z

@lilyevsky the relevant changes looks good, but your branch is out of sync with https://github.com/apache/flink-connector-opensearch, you need to sync + rebase

lilyevsky · 2023-02-15T23:50:08Z

@reta I see. Please disregard my previous message.
Do I sync + rebase on my desktop or I can do it in github directly?

lilyevsky · 2023-02-15T23:56:10Z

I think I can do it in github: https://docs.github.com/en/pull-requests/collaborating-with-pull-requests/working-with-forks/syncing-a-fork

lilyevsky · 2023-02-16T00:10:50Z

Done

reta · 2023-02-16T00:24:18Z

...ctor-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchSink.java

        this.hosts = checkNotNull(hosts);
        checkArgument(!hosts.isEmpty(), "Hosts cannot be empty.");
        this.emitter = checkNotNull(emitter);
        this.deliveryGuarantee = checkNotNull(deliveryGuarantee);
        this.buildBulkProcessorConfig = checkNotNull(buildBulkProcessorConfig);
        this.networkClientConfig = checkNotNull(networkClientConfig);
+        this.failureHandler = failureHandler;


Suggested change

this.failureHandler = failureHandler;

this.failureHandler = checkNotNull(failureHandler);

reta · 2023-02-16T00:24:42Z

...ensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilder.java

@@ -72,6 +72,7 @@
    private Integer connectionRequestTimeout;
    private Integer socketTimeout;
    private Boolean allowInsecure;
+    private FailureHandler failureHandler;


Suggested change

private FailureHandler failureHandler;

private FailureHandler failureHandler = DEFAULT_FAILURE_HANDLER;

reta · 2023-02-16T00:25:41Z

...or-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchWriter.java

@@ -343,7 +348,11 @@ private void extractFailures(BulkRequest request, BulkResponse response) {
        if (chainedFailures == null) {
            return;
        }
-        throw new FlinkRuntimeException(chainedFailures);
+        if (failureHandler == null) {


This null check is not needed anymore (the builder would supply the default handler), just failureHandler.onFailure(chainedFailures);

reta · 2023-02-16T00:27:09Z

Thanks @lilyevsky just a few minor comments, I think the only part that is missed are tests, and probably you may need to run mvn spotless:apply to apply the formatting, thank you!

lilyevsky · 2023-02-16T00:46:21Z

@reta Did all the changes you suggested, also ran the mvn spotless:apply .
Btw., my IntelliJ showed a list of warnings in places not relevant to my changes. Even one error (which is not really an error because it is in the comments), in OpensearchWriter.java:96.
Do you have any suggestions for the tests? As I mentioned before, I actually tested this in real environment; on the other hand, I understand that unit tests are very important to have.

reta · 2023-02-16T00:58:28Z

@reta Did all the changes you suggested, also ran the mvn spotless:apply .

👍 thank you!

Even one error (which is not really an error because it is in the comments), in OpensearchWriter.java:96.

Hm .. could you share the error please?

Do you have any suggestions for the tests?

Yes, sure, I think OpensearchWriterITCase could be a good candidate to add a test case with the simulated failure (fe we could pause / shutdown Opensearch container, just an idea). It may be a bit difficult, please let me know if help needed, thank you.

lilyevsky · 2023-02-16T01:41:47Z

The error is about this line:

@param bulkProcessorBuilderFactory configuring the {@link BulkProcessor}'s builder

Apparently that parameter is no longer present.
Also 10 warnings, like "Return value ... is never used" or "Exception ... is never thrown".
One warning is actually about failureHandler, missing description.

I actually can fix all of that if it is OK, then you can review.

Also will look into the tests tomorrow.

reta · 2023-02-16T01:49:04Z

I actually can fix all of that if it is OK, then you can review.

Thank you, cleanup would be much appreciated, I would suggest let's finish up with tests first so to have the scope finalized, and we could fix javadocs after. Thanks!

lilyevsky · 2023-02-16T01:53:27Z

Sure.
I will fix that bulkProcessorBuilderFactory thing, everything else is less important at the moment. I got that list of warnings while committing. When I just run 'analyze', I get even more stuff.

lilyevsky · 2023-02-16T17:43:45Z

@reta I fixed some warnings, pushed as a separate commit, so you can review.
Now looking at the tests. We need to somehow simulate the Opensearch server that can get the request and then send back the response. Is there currently such thing in the test suite? Is it what you do in flink-connector-opensearch-e2e-tests?
In any case, I think you would know better than I how to do it.

reta · 2023-02-16T18:14:10Z

@reta I fixed some warnings, pushed as a separate commit, so you can review. Now looking at the tests. We need to somehow simulate the Opensearch server that can get the request and then send back the response.

I shared the hints with you here #11 (comment)

Here is sample test case:

    @Test
    void testWriteErrorOnUpdate() throws Exception {
        final String index = "test-bulk-flush-with-error";
        final int flushAfterNActions = 1;
        final BulkProcessorConfig bulkProcessorConfig =
                new BulkProcessorConfig(flushAfterNActions, -1, -1, FlushBackoffType.NONE, 0, 0);

        try (final OpensearchWriter<Tuple2<Integer, String>> writer =
                createWriter(index, true, bulkProcessorConfig)) {
            // Trigger an error by updating non-existing document
            writer.write(Tuple2.of(1, "u" + buildMessage(1)), null);
            context.assertThatIdsAreNotWritten(index, 1);
        }
    }

You need to attach the failure handler and assert afterwards it has been called.

lilyevsky · 2023-02-16T20:03:37Z

@reta Trying to add the test. For some reason the tests from OpensearchWriterITCase class are not executed at all.
I run the command:

mvn clean test

in flink-connector-opensearch directory. I see that some tests from org.apache.flink.connector.opensearch.sink.OpensearchSinkBuilderTest are executed, but not from other 4 files in the same directory.
What could be wrong?

maventest.txt

reta · 2023-02-16T20:13:07Z

@lilyevsky OpensearchWriterITCase is integration test, mvn clean verify will run it

lilyevsky · 2023-02-16T20:29:09Z

Thanks @reta . One more step. It looks like I need to have a "valid Docker environment". I install the docker things:

sudo apt-get install docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin

What else do I need to do?

reta · 2023-02-16T21:07:15Z

@lilyevsky I assume you are on Ubuntu [1]

[1] https://docs.docker.com/engine/install/ubuntu/

lilyevsky · 2023-02-16T23:34:25Z

@reta I installed docker, the daemon is running.
However, the 'mvn verify' still fails with "OpensearchSinkITCase » IllegalState Could not find a valid Docker environment...." ....
I see that I have permissions problem:
docker run hello-world
docker: permission denied while trying to connect to the Docker daemon socket at unix:///var/run/docker.sock: Post "http://%2Fvar%2Frun%2Fdocker.sock/v1.24/containers/create": dial unix /var/run/docker.sock: connect: permission denied.

I think I can fix it. Will let you know.

lilyevsky · 2023-02-17T14:34:17Z

@reta Added the test. I think now it is good.
I also tested the test itself, by simulating the "bad" handler, making the test fail, just to make sure it checks it properly.

reta · 2023-02-17T15:04:25Z

...or-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchWriter.java

@@ -103,7 +107,8 @@
            BulkProcessorConfig bulkProcessorConfig,
            NetworkClientConfig networkClientConfig,
            SinkWriterMetricGroup metricGroup,
-            MailboxExecutor mailboxExecutor) {
+            MailboxExecutor mailboxExecutor,


Please update javadocs with failureHandler arg

reta · 2023-02-17T15:05:16Z

...nsearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java

+            return failed;
+        }
+
+        @java.lang.Override


Super nit, just @Override

reta · 2023-02-17T15:05:32Z

...nsearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java

@@ -238,19 +240,68 @@ void testCurrentSendTime() throws Exception {
        }
    }

+    private class TestHandler implements FailureHandler {


Probably private static is better

reta · 2023-02-17T15:05:54Z

...nsearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java

+            // Trigger an error by updating non-existing document
+            writer.write(Tuple2.of(1, "u" + buildMessage(1)), null);
+            context.assertThatIdsAreNotWritten(index, 1);
+            assertThat(testHandler.isFailed()).isEqualTo(true);


reta · 2023-02-17T15:07:32Z

...nsearch/src/test/java/org/apache/flink/connector/opensearch/sink/OpensearchWriterITCase.java

@@ -174,7 +175,8 @@ void testIncrementByteOutMetric() throws Exception {
                new BulkProcessorConfig(flushAfterNActions, -1, -1, FlushBackoffType.NONE, 0, 0);

        try (final OpensearchWriter<Tuple2<Integer, String>> writer =
-                createWriter(index, false, bulkProcessorConfig, metricGroup)) {
+                createWriter(


You probably don't need this change since createWriter has an overloaded version without failure handler

Agreed, it will be better not to touch that line. I did it because it needed to pass the metricGroup.
So I am fixing the situation by adding yet another constructor, with metricGroup but without failureHandler.
Also in the main constructor put the failureHandler at the end of the parameters list.

No need for constructor, createWriter is a method in this class

reta · 2023-02-17T15:09:56Z

...ctor-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/FailureHandler.java

+import java.io.Serializable;
+
+/** Handler to process failures. */
+@Public


@Public -> @PublicEvolving would be probably better (to leave some space to evolve the API)

reta · 2023-02-17T15:10:29Z

@lilyevsky a few minor comments, LGTM otherwise!

lilyevsky · 2023-02-17T16:36:00Z

@reta I committed the changes you suggested.

reta · 2023-02-17T16:45:16Z

...or-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchWriter.java

    }

    @Override
-    public void write(IN element, Context context) throws IOException, InterruptedException {
+    public void write(IN element, Context context) throws InterruptedException {


Please revert these changes, they violate the SinkWriter interface

[1] https://github.com/apache/flink/blob/master/flink-core/src/main/java/org/apache/flink/api/connector/sink2/SinkWriter.java#L41

@lilyevsky could you please address this and this comment? thank you

@reta No problem, could you please clarify: you want me to put back the IOException to both write and flush methods, correct? Also, I am not sure what you mean by addressing this . Is it about removing the "@SuppressWarnings("All")"?
Please confirm.

@lilyevsky correct, the throws IOException, InterruptedException is the right part of the signature: the first comment is about write, the second is about flush, thank you

@reta Done. Also removed the line with warnings suppressions that I added at some point.

reta · 2023-02-17T16:45:25Z

...or-opensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchWriter.java

@@ -134,7 +141,7 @@ public void write(IN element, Context context) throws IOException, InterruptedEx
    }

    @Override
-    public void flush(boolean endOfInput) throws IOException, InterruptedException {
+    public void flush(boolean endOfInput) throws InterruptedException {


Same, please revert

reta · 2023-02-17T16:47:06Z

...ensearch/src/main/java/org/apache/flink/connector/opensearch/sink/OpensearchSinkBuilder.java

@@ -148,6 +150,7 @@ public OpensearchSinkBuilder<IN> setBulkFlushMaxActions(int numMaxActions) {
     * @param maxSizeMb the maximum size of buffered actions, in mb.
     * @return this builder
     */
+    @SuppressWarnings("UnusedReturnValue")


I don't understand why these suppression are needed?

I put it because compiler gave me a warning about it. I can remove that suppression if you prefer.

Please remove, thank you

Hi @reta , please let me know what are the plans for merging.

Hi @reta , please let me know what are the plans for merging.

Thanks @lilyevsky , LGTM, asked Flink committers to take I look (I do not have commit rights)

Thanks @reta

reta · 2023-03-04T16:04:24Z

@MartijnVisser @zentol could you please take a look guys? thank you!

lilyevsky · 2023-07-06T18:45:35Z

I see since I finalized this request there were some other changes in main branch, and so there are conflicts.
I am going to rebase my branch and hopefully resolve the conflicts this way.
After I do that, can we merge my request sooner rather than later? Otherwise there might be other conflicts again.

lilyevsky · 2023-07-06T20:37:09Z

I actually resolved the conflicts in the web editor, and updated my branch.
Can somebody review this?

reta · 2023-07-06T21:38:22Z

@MartijnVisser @snuyanzin please help us here :-)

lilyevsky · 2023-07-06T21:52:28Z

Actually, hold on for now. I was not able to compile it in my work environment, now tried in my home desktop and found some errors. Let me fix them. Will let you know.
I may still need to rebase my branch on apache:main, or merge it in. Which way it is easier?
Or maybe I will just repeat all my changes on top of the latest upstream, could be easier to make it compile clean.

…in branch

lilyevsky · 2023-07-06T23:03:11Z

I made it clean now, you can review and maybe merge already.
I got the latest from apache:main branch and re-applied all my changes on top of it, so you can see now it is just one commit.
I was able to locally build it.

lilyevsky · 2023-07-07T17:13:41Z

So you can merge it now?
After you do, I am going to build locally the latest version and do my own tests.

boring-cyborg · 2023-07-11T16:37:06Z

Awesome work, congrats on your first merged pull request!

boring-cyborg bot added the component=Connectors/Opensearch label Feb 15, 2023

lilyevsky force-pushed the main branch 2 times, most recently from 80ea673 to ff041bd Compare February 15, 2023 23:26

boring-cyborg bot added the component=BuildSystem label Feb 15, 2023

lilyevsky force-pushed the main branch 2 times, most recently from 79222a1 to 1c4eda0 Compare February 16, 2023 00:09

reta reviewed Feb 16, 2023

View reviewed changes

reta reviewed Feb 17, 2023

View reviewed changes

reta approved these changes Mar 4, 2023

View reviewed changes

[FLINK-30998] Apply adding failureHandler on top of current apache:ma…

a6f4639

…in branch

lilyevsky force-pushed the main branch from ce344d7 to a6f4639 Compare July 6, 2023 22:59

MartijnVisser requested a review from reta July 7, 2023 12:07

reta approved these changes Jul 7, 2023

View reviewed changes

MartijnVisser merged commit d853e3d into apache:main Jul 11, 2023
5 checks passed

reta mentioned this pull request Dec 6, 2023

[FLINK-32028][connectors/elasticsearch] Allow customising bulk failure handling apache/flink-connector-elasticsearch#83

Merged

	this.failureHandler = failureHandler;
	this.failureHandler = checkNotNull(failureHandler);

	private FailureHandler failureHandler;
	private FailureHandler failureHandler = DEFAULT_FAILURE_HANDLER;

[FLINK-30998] Add optional exception handler to flink-connector-opensearch #11

[FLINK-30998] Add optional exception handler to flink-connector-opensearch #11

Conversation

lilyevsky commented Feb 15, 2023

reta commented Feb 15, 2023

lilyevsky commented Feb 15, 2023 • edited

lilyevsky commented Feb 15, 2023

lilyevsky commented Feb 16, 2023

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

reta commented Feb 16, 2023

lilyevsky commented Feb 16, 2023

reta commented Feb 16, 2023

lilyevsky commented Feb 16, 2023

reta commented Feb 16, 2023

lilyevsky commented Feb 16, 2023

lilyevsky commented Feb 16, 2023

reta commented Feb 16, 2023

lilyevsky commented Feb 16, 2023

reta commented Feb 16, 2023

lilyevsky commented Feb 16, 2023

reta commented Feb 16, 2023

lilyevsky commented Feb 16, 2023 • edited

lilyevsky commented Feb 17, 2023

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

lilyevsky Feb 17, 2023 • edited

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

reta commented Feb 17, 2023

lilyevsky commented Feb 17, 2023

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

reta commented Mar 4, 2023

lilyevsky commented Jul 6, 2023

lilyevsky commented Jul 6, 2023

reta commented Jul 6, 2023

lilyevsky commented Jul 6, 2023 • edited

lilyevsky commented Jul 6, 2023

lilyevsky commented Jul 7, 2023

boring-cyborg bot commented Jul 11, 2023

lilyevsky commented Feb 15, 2023 •

edited

lilyevsky commented Feb 16, 2023 •

edited

lilyevsky Feb 17, 2023 •

edited

lilyevsky commented Jul 6, 2023 •

edited