zentol commented on a change in pull request #15105:
URL: https://github.com/apache/flink/pull/15105#discussion_r591399729
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/registration/RetryingRegistrationTest.java
##########
@@ -373,10 +386,10 @@ public String getCorrelationId() {
static final long INITIAL_TIMEOUT = 20;
static final long MAX_TIMEOUT = 200;
static final long DELAY_ON_ERROR = 200;
- static final long DELAY_ON_DECLINE = 200;
+ static final long DELAY_ON_FAILURE = 200;
Review comment:
~~the previous name is more in line with the naming in the
`RetryingRegistrationConfiguration`.~~
I see, it makes more sense with the subsequent commit.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/registration/ManualResponseTestRegistrationGateway.java
##########
@@ -0,0 +1,88 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package org.apache.flink.runtime.registration;
+
+import org.apache.flink.runtime.rpc.TestingGatewayBase;
+import org.apache.flink.util.Preconditions;
+
+import java.util.UUID;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.LinkedBlockingQueue;
+
+/** Mock gateway for {@link RegistrationResponse}. */
+public class ManualResponseTestRegistrationGateway extends TestingGatewayBase
+ implements TestRegistrationGateway {
+
+ private final BlockingQueue<RegistrationCall> invocations;
+
+ private final RegistrationResponse[] responses;
+
+ private int pos;
+
+ public ManualResponseTestRegistrationGateway(RegistrationResponse...
responses) {
+ Preconditions.checkArgument(responses != null && responses.length > 0);
+
+ this.invocations = new LinkedBlockingQueue<>();
+ this.responses = responses;
+ }
+
+ // ------------------------------------------------------------------------
+
+ @Override
+ public CompletableFuture<RegistrationResponse> registrationCall(UUID
leaderId, long timeout) {
+ invocations.add(new RegistrationCall(leaderId, timeout));
+
+ RegistrationResponse response = responses[pos];
+ if (pos < responses.length - 1) {
Review comment:
Why are we not failing hard if this is violated? It is quite subtle
behavior that the last response may be returned an infinite number of times.
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/registration/RegistrationResponse.java
##########
@@ -42,30 +44,42 @@ public String toString() {
//
----------------------------------------------------------------------------
- /** A rejected (declined) registration. */
- public static final class Decline extends RegistrationResponse {
+ /** A registration failure. */
+ public static final class Failure extends RegistrationResponse {
private static final long serialVersionUID = 1L;
- /** The rejection reason. */
- private final String reason;
+ /** The failure reason. */
+ private final SerializedThrowable reason;
/**
- * Creates a new rejection message.
+ * Creates a new failure message.
*
- * @param reason The reason for the rejection.
+ * @param reason The reason for the failure.
*/
- public Decline(String reason) {
- this.reason = reason != null ? reason : "(unknown)";
+ public Failure(Throwable reason) {
+ this.reason = new SerializedThrowable(reason);
}
- /** Gets the reason for the rejection. */
- public String getReason() {
+ /** Gets the reason for the failure. */
+ public SerializedThrowable getReason() {
return reason;
}
@Override
public String toString() {
- return "Registration Declined (" + reason + ')';
+ return "Registration Failure (" + reason + ')';
+ }
+ }
+
+ //
----------------------------------------------------------------------------
+
+ /** A rejected (declined) registration. */
+ public static class Rejection extends RegistrationResponse {
Review comment:
Why does this not have a cause or description for what the problem was?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
##########
@@ -223,16 +229,22 @@ private void register(final G gateway, final int attempt,
final long timeoutMill
if (result instanceof
RegistrationResponse.Success) {
// registration successful!
S success = (S) result;
-
completionFuture.complete(Tuple2.of(gateway, success));
+ completionFuture.complete(
+
RetryingRegistrationResult.success(
+ gateway, success));
+ } else if (result instanceof
RegistrationResponse.Rejection) {
+ R rejection = (R) result;
+ completionFuture.complete(
+
RetryingRegistrationResult.rejection(rejection));
Review comment:
shouldn't this be logged?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/taskexecutor/TaskExecutor.java
##########
@@ -2064,11 +2101,19 @@ public void jobManagerLostLeadership(final JobID jobId,
final JobMasterId jobMas
public void handleError(Throwable throwable) {
onFatalError(throwable);
}
+
+ @Override
+ public void jobManagerRejectedRegistration(
+ JobID jobId, String targetAddress, JMTMRegistrationRejection
rejection) {
+ runAsync(() -> handleRejectedJobManagerConnection(jobId,
targetAddress, rejection));
Review comment:
_should_ it actually clean up everything? Why is it not just asking for
the actual leader, and only failing if that times out?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/registration/RetryingRegistration.java
##########
@@ -223,16 +229,22 @@ private void register(final G gateway, final int attempt,
final long timeoutMill
if (result instanceof
RegistrationResponse.Success) {
// registration successful!
S success = (S) result;
-
completionFuture.complete(Tuple2.of(gateway, success));
+ completionFuture.complete(
+
RetryingRegistrationResult.success(
+ gateway, success));
+ } else if (result instanceof
RegistrationResponse.Rejection) {
+ R rejection = (R) result;
+ completionFuture.complete(
+
RetryingRegistrationResult.rejection(rejection));
} else {
// registration refused or unknown
Review comment:
outdated?
##########
File path:
flink-runtime/src/main/java/org/apache/flink/runtime/registration/RegistrationResponse.java
##########
@@ -42,30 +44,42 @@ public String toString() {
//
----------------------------------------------------------------------------
- /** A rejected (declined) registration. */
- public static final class Decline extends RegistrationResponse {
+ /** A registration failure. */
+ public static final class Failure extends RegistrationResponse {
private static final long serialVersionUID = 1L;
- /** The rejection reason. */
- private final String reason;
+ /** The failure reason. */
+ private final SerializedThrowable reason;
/**
- * Creates a new rejection message.
+ * Creates a new failure message.
*
- * @param reason The reason for the rejection.
+ * @param reason The reason for the failure.
*/
- public Decline(String reason) {
- this.reason = reason != null ? reason : "(unknown)";
+ public Failure(Throwable reason) {
+ this.reason = new SerializedThrowable(reason);
}
- /** Gets the reason for the rejection. */
- public String getReason() {
+ /** Gets the reason for the failure. */
+ public SerializedThrowable getReason() {
return reason;
}
@Override
public String toString() {
- return "Registration Declined (" + reason + ')';
+ return "Registration Failure (" + reason + ')';
+ }
+ }
+
+ //
----------------------------------------------------------------------------
+
+ /** A rejected (declined) registration. */
Review comment:
We should clarify in what circumstances a Rejection or Failure should be
returned.
##########
File path:
flink-runtime/src/test/java/org/apache/flink/runtime/registration/RegisteredRpcConnectionTest.java
##########
@@ -102,12 +98,17 @@ public void testRpcConnectionFailures() throws Exception {
final String testRpcConnectionEndpointAddress =
"<TestRpcConnectionEndpointAddress>";
final UUID leaderId = UUID.randomUUID();
- // gateway that upon calls Throw an exception
- TestRegistrationGateway testGateway =
mock(TestRegistrationGateway.class);
final RuntimeException registrationException =
new RuntimeException(connectionFailureMessage);
- when(testGateway.registrationCall(any(UUID.class), anyLong()))
- .thenThrow(registrationException);
+
+ // gateway that upon calls Throw an exception
Review comment:
> // gateway that upon registration calls throws an exception
----------------------------------------------------------------
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org
