[FLINK-14807][runtime][client] Introduce communication through REST API between the clients and the operator coordinators #12037

tsreaper · 2020-05-08T11:12:27Z

What is the purpose of the change

This PR is part of the FLINK-14807 which is going to introduce a collecting method for tables. See here for the whole design document.

To allow the clients to receive results from the coordinator, which acts as a proxy between the clients and the sinks, a communication protocol should be introduced between the clients and the coordinators. This PR introduce this new communication protocol through the REST API.

Note that as coordinators are internally used only, we're not going to expose this REST API to the user, thus no user document is provided.

Brief change log

Introduce a REST API to allow the communication between clients and coordinators.

Verifying this change

The changes can be verified by running the added unit tests.

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): no
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: yes (JobManager)
The S3 file system connector: no

Documentation

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

flinkbot · 2020-05-08T11:14:48Z

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 ab6e6c9 (Fri May 08 11:14:48 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-05-08T15:28:15Z

CI report:

d515b32 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

tsreaper · 2020-05-09T07:43:05Z

cc @becketqin @StephanEwen @godfreyhe please help for the review, thanks a lot~

godfreyhe

Thanks for the contribution @tsreaper. I left a few of comments

godfreyhe · 2020-05-11T02:45:24Z

...t/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java

+
+ assertThat(future, futureFailedWith(IllegalArgumentException.class));
+ }
+


add a test for scenario: "operator id not found"

godfreyhe · 2020-05-11T02:48:44Z

flink-runtime/src/main/java/org/apache/flink/runtime/scheduler/SchedulerBase.java

+ ((CoordinationResponser) coordinator).handleCoordinationRequest(request));
+ } else if (coordinator != null) {
+ return FutureUtils.completedExceptionally(
+ new IllegalArgumentException("Coordinator of operator " + operator + " cannot handle client event"));


I think we should throw FlinkException, this situation is the same as the operation does not exist

godfreyhe · 2020-05-11T02:54:47Z

flink-clients/src/main/java/org/apache/flink/client/program/rest/RestClusterClient.java

+ .getSerializedCoordinationResponse()
+ .deserializeValue(getClass().getClassLoader());
+ } catch (IOException | ClassNotFoundException e) {
+ throw new RuntimeException("Failed to deserialize coordination response", e);


throw CompletionException like getAccumulators method

godfreyhe · 2020-05-11T03:01:54Z

...ime/src/main/java/org/apache/flink/runtime/operators/coordination/CoordinationResponser.java

+ * Coordinator interface which can handle {@link CoordinationRequest}s
+ * and response with {@link CoordinationResponse}s to the client.
+ */
+public interface CoordinationResponser {


rename to CoordinationHandler ? similar to OperatorEventHandler

CoordinationRequestHandler seems to be better

godfreyhe · 2020-05-11T03:06:33Z

...ime/src/main/java/org/apache/flink/runtime/operators/coordination/CoordinationRequester.java

+ * Client interface which sends out a {@link CoordinationRequest} and
+ * expects for a {@link CoordinationResponse} from a {@link OperatorCoordinator}.
+ */
+public interface CoordinationRequester {


rename to CoordinationRequestGateway? similar to OperatorEventGateway

KurtYoung · 2020-05-12T04:04:29Z

...rc/main/java/org/apache/flink/runtime/operators/coordination/CoordinationRequestGateway.java

+ * Client interface which sends out a {@link CoordinationRequest} and
+ * expects for a {@link CoordinationResponse} from a {@link OperatorCoordinator}.
+ */
+public interface CoordinationRequestGateway {


I'm not sure this interface is necessary, it seems we can add the method to JobClient. You can see almost each time when a class implements JobClient, it will also implement this interface

Coordinators are used internally so we do not want to expose this to the users. As JobClient is a public interface, we prefer not to modify it.

KurtYoung · 2020-05-12T04:18:35Z

...n/java/org/apache/flink/runtime/rest/handler/job/coordination/ClientCoordinationHandler.java

+ try {
+ return new ClientCoordinationResponseBody(new SerializedValue<>(coordinationResponse));
+ } catch (IOException e) {
+ throw new RuntimeException("Failed to construct response body", e);


throw RestHandlerException instead? Will the RuntimeException crash the rest server?

It won't. See JobAccumulatorsHandler for similar behaviors.

Throwing RuntimeExceptions is an implementation error. REST handlers should always throw a RestHandlerException, as only then a proper error is returned to the user, instead of a generic 500 Internal Server Error.

tsreaper · 2020-05-13T02:13:29Z

@flinkbot run azure

KurtYoung · 2020-05-13T03:09:34Z

flink-runtime/src/main/java/org/apache/flink/runtime/jobmaster/JobMasterGateway.java

+ * if the task is not running, or no operator/coordinator exists for the given ID,
+ * or the coordinator cannot handle client events.
+ */
+ CompletableFuture<CoordinationResponse> deliverCoordinationRequestToCoordinator(


I think this should add to JobMasterOperatorEventGateway

No. JobMasterOperatorEventGateway is for OperatorEvents. CoordinationRequest and CoordinationResponse are different from OperatorEvent.

tillrohrmann

Thanks a lot for creating this PR @tsreaper. Nice work! The changes look good to me. I had a couple of minor comments which would be good to be addressed before we merge this PR. Once this is done and AZP gives green light, we can merge it.

tillrohrmann · 2020-05-13T15:30:29Z

...n/java/org/apache/flink/runtime/rest/handler/job/coordination/ClientCoordinationHandler.java

+ return responseFuture.thenApply(
+ coordinationResponse -> {
+ try {
+ return new ClientCoordinationResponseBody(new SerializedValue<>(coordinationResponse));


Why are we wrapping coordinationResponse into a SerializedValue here? If it is in order to support user code classes, then it should not work because ClientCoordinationHandler should not have access to the user code class loader.

The wrapping would have to happen on a component which has access to the user code classes.

Because CoordinationResponse is an object and should be serialized in order to be passed back to the client through the REST API. Is there a more proper way to achieve this?

Makes sense. I guess we will use another layer of SerializedValue if the user requests a user code object from the OperatorCoordinator which is then initialized on the JM.

tillrohrmann · 2020-05-13T15:35:55Z

...t/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java

@@ -163,6 +165,56 @@ public void taskTaskManagerFailuresAreReportedBack() throws Exception {
 assertThat(result, futureFailedWith(TestException.class));
 }

+ @Test
+ @SuppressWarnings("unchecked")
+ public void testDeliveringClientRequestToResponser() throws Exception {


Suggested change

public void testDeliveringClientRequestToResponser() throws Exception {

public void testDeliveringClientRequestToResponder() throws Exception {

This should be testDeliveringClientRequestToRequestHandler now as the name of the interface is changed to CoordinationRequestHandler.

tillrohrmann · 2020-05-13T15:36:18Z

...t/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java

+ }
+
+ @Test(expected = FlinkException.class)
+ public void testDeliveringClientRequestToNonResponser() throws Exception {


Suggested change

public void testDeliveringClientRequestToNonResponser() throws Exception {

public void testDeliveringClientRequestToNonResponder() throws Exception {

This should be testDeliveringClientRequestToNonRequestHandler now as the name of the interface is changed to CoordinationRequestHandler.

tillrohrmann · 2020-05-13T18:55:33Z

...t/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java

+ } catch (FlinkException e) {
+ Assert.assertTrue(e.getMessage().contains("cannot handle client event"));
+ throw e;
+ }


I think we could use CommonTestUtils.assertThrows here.

tillrohrmann · 2020-05-13T18:56:00Z

...t/java/org/apache/flink/runtime/operators/coordination/OperatorCoordinatorSchedulerTest.java

+ Assert.assertTrue(e.getMessage().contains("does not exist"));
+ throw e;
+ }


same here with CommonTestUtils.assertThrows

zentol

Are we intentionally not regenerating the REST documentation?

tsreaper · 2020-05-14T02:07:34Z

Are we intentionally not regenerating the REST documentation?

Yes. As coordinators are used only internally, we're not going to expose this to the users so no user documentation is updated.

I've updated RestAPIDocGenerator in FLINK-17680 and #12141 , please take a look.

tillrohrmann · 2020-05-14T07:07:27Z

I like the idea to have an internal REST API which we don't expose to the user because it is not intended for public use.

tillrohrmann · 2020-05-14T09:41:53Z

@flinkbot run azure

tsreaper · 2020-05-14T09:47:36Z

@flinkbot run azure

It has already finished, see here, all tests have passed except e2e tests which are cancelled.

tillrohrmann · 2020-05-14T09:58:26Z

True. Thanks for the information.

zentol

I'm fine with having internal REST API's, so if #12141 is merged beforehand and the headers are marked accordingly then I'm fine with it.

tillrohrmann · 2020-05-14T14:12:08Z

@tsreaper I have merged #12141. You could now rebase this PR on the current master and add the @ExcludeFromDocumentation annotation.

…PI between the clients and the operator coordinators

…Future in case of costly handling

tsreaper · 2020-05-14T14:33:11Z

@tillrohrmann @zentol I've rebased onto master and marked ClientCoordinationHeaders with @ExcludeFromDocumentation, please take another look. Thanks.

tillrohrmann · 2020-05-14T16:06:02Z

Thanks @tsreaper. Merging this PR once AZP gives green light.

KurtYoung · 2020-05-14T23:44:19Z

AZP passed, I'm merging this then I can start to review another PR based on this.

rmetzger added review=description? component=TableSQL/API labels May 8, 2020

godfreyhe reviewed May 11, 2020

View reviewed changes

This was referenced May 11, 2020

[FLINK-17734][streaming] Add specialized collecting sink function #12069

Closed

[FLINK-17735][streaming] Add specialized collecting iterator #12073

Merged

KurtYoung reviewed May 12, 2020

View reviewed changes

KurtYoung reviewed May 13, 2020

View reviewed changes

tillrohrmann self-assigned this May 13, 2020

tillrohrmann approved these changes May 13, 2020

View reviewed changes

zentol requested changes May 13, 2020

View reviewed changes

zentol reviewed May 14, 2020

View reviewed changes

tsreaper added 7 commits May 14, 2020 22:29

[FLINK-14807][runtime][client] Introduce communication through REST A…

18e3fa4

…PI between the clients and the operator coordinators

[fix] Update REST API test resource

4e0dcdb

[fix] Fix godfrey's comments

2f8b362

[fix] Fix type of exception thrown in REST handler

d381e1d

[fix] Change return type of CoordinationRequestHandler to Completable…

2fd031d

…Future in case of costly handling

[fix] Fix unit test naming and assertion

7166470

[fix] Mark ClientCoordinationHeaders with @ExcludeFromDocumentation

d515b32

KurtYoung merged commit 9fe920f into apache:master May 14, 2020


		assertThat(future, futureFailedWith(IllegalArgumentException.class));
		}

	public void testDeliveringClientRequestToResponser() throws Exception {
	public void testDeliveringClientRequestToResponder() throws Exception {

[FLINK-14807][runtime][client] Introduce communication through REST API between the clients and the operator coordinators #12037

[FLINK-14807][runtime][client] Introduce communication through REST API between the clients and the operator coordinators #12037

Conversation

tsreaper commented May 8, 2020 • edited Loading

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 May 8, 2020

Automated Checks

Review Progress

flinkbot commented May 8, 2020 • edited Loading

CI report:

tsreaper commented May 9, 2020

godfreyhe 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

tsreaper May 12, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tsreaper commented May 13, 2020

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tillrohrmann left a comment • edited Loading

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

tsreaper May 14, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

tsreaper May 14, 2020 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

zentol left a comment

Choose a reason for hiding this comment

tsreaper commented May 14, 2020 • edited Loading

tillrohrmann commented May 14, 2020

tillrohrmann commented May 14, 2020

tsreaper commented May 14, 2020 • edited Loading

tillrohrmann commented May 14, 2020

zentol left a comment

Choose a reason for hiding this comment

tillrohrmann commented May 14, 2020

tsreaper commented May 14, 2020

tillrohrmann commented May 14, 2020

KurtYoung commented May 14, 2020

tsreaper commented May 8, 2020 •

edited

Loading

flinkbot commented May 8, 2020 •

edited

Loading

tsreaper May 12, 2020 •

edited

Loading

tillrohrmann left a comment •

edited

Loading

tsreaper May 14, 2020 •

edited

Loading

tsreaper May 14, 2020 •

edited

Loading

tsreaper commented May 14, 2020 •

edited

Loading

tsreaper commented May 14, 2020 •

edited

Loading