[FLINK-17623][Connectors/Elasticsearch] Support user resource cleanup in Elasticsearch sink #12619

yun-wang · 2020-06-12T00:40:02Z

What is the purpose of the change

This pull request supports user resource cleanup in Elasticsearch sink.

Brief change log

Expose a default close method in ElasticsearchSinkFunction.

Verifying this change

This change added tests and can be verified as follows:

Added a unit test that both open() and close() methods are invoked in the Elasticsearch sink as expected.

Does this pull request potentially affect one of the following parts:

Dependencies (does it add or upgrade a dependency): (no)
The public API, i.e., is any changed class annotated with @Public(Evolving): (yes)
The serializers: (no)
The runtime per-record code paths (performance sensitive): (no)
Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn/Mesos, ZooKeeper: (no)
The S3 file system connector: (no)

Documentation

Does this pull request introduce a new feature? (yes)
If yes, how is the feature documented? (JavaDocs)

flinkbot · 2020-06-12T00:42:29Z

Thanks a lot for your contribution to the Apache Flink project. I'm the @flinkbot. I help the community
to review your pull request. We will use this comment to track the progress of the review.

Automated Checks

Last check on commit a73bb36 (Fri Jun 12 00:42:29 UTC 2020)

Warnings:

No documentation files were touched! Remember to keep the Flink docs up to date!

_{Mention the bot in a comment to re-run the automated checks.}

Review Progress

❓ 1. The [description] looks good.
❓ 2. There is [consensus] that the contribution should go into to Flink.
❓ 3. Needs [attention] from.
❓ 4. The change fits into the overall [architecture].
❓ 5. Overall code [quality] is good.

Please see the Pull Request Review Guide for a full explanation of the review process.

The Bot is tracking the review progress through labels. Labels are applied according to the order of the review items. For consensus, approval by a Flink committer of PMC member is required

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot approve description to approve one or more aspects (aspects: description, consensus, architecture and quality)
@flinkbot approve all to approve all aspects
@flinkbot approve-until architecture to approve everything until architecture
@flinkbot attention @username1 [@username2 ..] to require somebody's attention
@flinkbot disapprove architecture to remove an approval you gave earlier

flinkbot · 2020-06-12T01:05:58Z

CI report:

9249535 Azure: SUCCESS

Bot commands

The @flinkbot bot supports the following commands:

@flinkbot run travis re-run the last Travis build
@flinkbot run azure re-run the last Azure build

KarmaGYZ

Thanks for the PR, @yun-wang . I left two minor comments.

KarmaGYZ · 2020-06-12T02:01:04Z

...main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkFunction.java

+	/**
+	 * Tear-down method for the function. It is called when the sink closes.
+	 */
+	default void close () throws Exception {}


It's a bit weird that the open function does not allow to throw an exception. I admit it is out of the scope of this PR though.

I can update open to default void open() throws Exception {} real quick. It's a backward compatible change and should have no user impact.

Sounds good to me.

KarmaGYZ · 2020-06-12T02:04:22Z

...test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java

+		ElasticsearchSinkFunction<String> sinkFunction = (ElasticsearchSinkFunction<String>) mock(ElasticsearchSinkFunction.class);
+		final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>(
+				new HashMap<>(), sinkFunction, new DummyRetryFailureHandler());
+
+		sink.open(mock(Configuration.class));
+		sink.close();
+
+		verify(sinkFunction, times(1)).open();
+		verify(sinkFunction, times(1)).close();
+		verifyNoMoreInteractions(sinkFunction);


The community tends to avoid mock if possible. In this case, I think we could add TestingElasticsearchSinkFunction or just an anonymous class.

Sure I can update the test.

Out of curiosity, any particular reason to avoid mock whenever possible? In this case adding a simple implementation class would serve the exact same purpose as the mock, only lengthier.

I think this reflects how well our design adheres to the principles of dependency injection. If our design is good enough, we could directly use the constructor in the test.

… in Elasticsearch sink

…en to throw an exception

KarmaGYZ · 2020-06-15T02:20:59Z

...test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java

+		final DummyElasticsearchSink<String> sink = new DummyElasticsearchSink<>(
+				new HashMap<>(), sinkFunction, new DummyRetryFailureHandler());
+
+		sink.open(mock(Configuration.class));


Suggested change

sink.open(mock(Configuration.class));

sink.open(new Configuration()));

KarmaGYZ · 2020-06-15T02:22:49Z

...test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java

@@ -586,6 +600,27 @@ public void process(String element, RuntimeContext ctx, RequestIndexer indexer)
 		}
 	}

+	private static class SimpleClosableSinkFunction<String> implements ElasticsearchSinkFunction<String> {


Maybe TestingClosableSinkFunction? Not quite sure about this.

KarmaGYZ

Thanks for addressing my comments, @yun-wang . It now LGTM. Just two minor comments you could take a look.

KarmaGYZ · 2020-06-15T02:26:18Z

@wuchong Could you help to review and merge it?

wuchong

Thanks for the work @yun-wang and reviewing @KarmaGYZ .

LGTM.

…searchSinkFunction This closes apache#12619

rmetzger added the review=description? label Jun 12, 2020

rmetzger added the component=Connectors/ElasticSearch label Jun 12, 2020

KarmaGYZ reviewed Jun 12, 2020

View reviewed changes

yun-wang added 2 commits June 12, 2020 13:23

[FLINK-17623][Connectors/Elasticsearch] Support user resource cleanup…

3e4050c

… in Elasticsearch sink

[hotfix][Connectors/Elasticsearch] Allow ElasticsearchSinkFunction#op…

9249535

…en to throw an exception

yun-wang force-pushed the FLINK-17623 branch from a73bb36 to 9249535 Compare June 12, 2020 20:32

KarmaGYZ reviewed Jun 15, 2020

View reviewed changes

KarmaGYZ approved these changes Jun 15, 2020

View reviewed changes

wuchong approved these changes Jun 15, 2020

View reviewed changes

wuchong closed this in 57992c9 Jun 15, 2020

bigdata-ny pushed a commit to bigdata-ny/flink that referenced this pull request Jun 19, 2020

[FLINK-17623][elasticsearch] Support user resource cleanup in Elastic…

9afe9eb

…searchSinkFunction This closes apache#12619

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

[FLINK-17623][Connectors/Elasticsearch] Support user resource cleanup in Elasticsearch sink #12619

[FLINK-17623][Connectors/Elasticsearch] Support user resource cleanup in Elasticsearch sink #12619

yun-wang commented Jun 12, 2020

flinkbot commented Jun 12, 2020

flinkbot commented Jun 12, 2020 •

edited

KarmaGYZ left a comment

KarmaGYZ Jun 12, 2020

yun-wang Jun 12, 2020

KarmaGYZ Jun 15, 2020

KarmaGYZ Jun 12, 2020 •

edited

yun-wang Jun 12, 2020

KarmaGYZ Jun 15, 2020

KarmaGYZ Jun 15, 2020

KarmaGYZ Jun 15, 2020

KarmaGYZ left a comment

KarmaGYZ commented Jun 15, 2020

wuchong left a comment

	sink.open(mock(Configuration.class));
	sink.open(new Configuration()));

[FLINK-17623][Connectors/Elasticsearch] Support user resource cleanup in Elasticsearch sink #12619

[FLINK-17623][Connectors/Elasticsearch] Support user resource cleanup in Elasticsearch sink #12619

Conversation

yun-wang commented Jun 12, 2020

What is the purpose of the change

Brief change log

Verifying this change

Does this pull request potentially affect one of the following parts:

Documentation

flinkbot commented Jun 12, 2020

Automated Checks

Review Progress

flinkbot commented Jun 12, 2020 • edited

CI report:

KarmaGYZ left a comment

Choose a reason for hiding this comment

KarmaGYZ Jun 12, 2020

Choose a reason for hiding this comment

yun-wang Jun 12, 2020

Choose a reason for hiding this comment

KarmaGYZ Jun 15, 2020

Choose a reason for hiding this comment

KarmaGYZ Jun 12, 2020 • edited

Choose a reason for hiding this comment

yun-wang Jun 12, 2020

Choose a reason for hiding this comment

KarmaGYZ Jun 15, 2020

Choose a reason for hiding this comment

KarmaGYZ Jun 15, 2020

Choose a reason for hiding this comment

KarmaGYZ Jun 15, 2020

Choose a reason for hiding this comment

KarmaGYZ left a comment

Choose a reason for hiding this comment

KarmaGYZ commented Jun 15, 2020

wuchong left a comment

Choose a reason for hiding this comment

flinkbot commented Jun 12, 2020 •

edited

KarmaGYZ Jun 12, 2020 •

edited