You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/08/21 07:19:54 UTC
[flink] branch master updated (ed4937cbc0e -> b5b1340f91d)
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
from ed4937cbc0e [hotfix] Fix the deadlock of DiskIOScheduler#sortScheduledReaders of Hybrid Shuffle
new 1c76850ec7a [hotfix][test] Migrates RestClientTest to JUnit5/AssertJ
new b5b1340f91d [FLINK-32751][runtime] Refactors CollectSinkOperatorCoordinator to enable case-specific logging
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../apache/flink/runtime/rest/RestClientTest.java | 140 +++++++++------------
.../collect/CollectSinkOperatorCoordinator.java | 140 ++++++++++++---------
2 files changed, 144 insertions(+), 136 deletions(-)
[flink] 01/02: [hotfix][test] Migrates RestClientTest to JUnit5/AssertJ
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 1c76850ec7a23b5bee94b96689bcdb06b8e29457
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Tue Aug 8 17:56:44 2023 +0200
[hotfix][test] Migrates RestClientTest to JUnit5/AssertJ
---
.../apache/flink/runtime/rest/RestClientTest.java | 140 +++++++++------------
1 file changed, 62 insertions(+), 78 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java
index f85117cfb50..9423335738b 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/rest/RestClientTest.java
@@ -20,6 +20,7 @@ package org.apache.flink.runtime.rest;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
+import org.apache.flink.core.testutils.FlinkAssertions;
import org.apache.flink.core.testutils.OneShotLatch;
import org.apache.flink.runtime.rest.messages.EmptyMessageParameters;
import org.apache.flink.runtime.rest.messages.EmptyRequestBody;
@@ -27,10 +28,8 @@ import org.apache.flink.runtime.rest.messages.EmptyResponseBody;
import org.apache.flink.runtime.rest.messages.RuntimeMessageHeaders;
import org.apache.flink.runtime.rest.versioning.RuntimeRestAPIVersion;
import org.apache.flink.testutils.TestingUtils;
-import org.apache.flink.testutils.executor.TestExecutorResource;
-import org.apache.flink.util.ExceptionUtils;
+import org.apache.flink.testutils.executor.TestExecutorExtension;
import org.apache.flink.util.NetUtils;
-import org.apache.flink.util.TestLogger;
import org.apache.flink.util.concurrent.Executors;
import org.apache.flink.util.function.CheckedSupplier;
@@ -41,10 +40,9 @@ import org.apache.flink.shaded.netty4.io.netty.channel.SelectStrategy;
import org.apache.flink.shaded.netty4.io.netty.channel.SelectStrategyFactory;
import org.apache.flink.shaded.netty4.io.netty.handler.codec.http.HttpResponseStatus;
-import org.junit.Assert;
-import org.junit.ClassRule;
-import org.junit.Test;
-import org.junit.function.ThrowingRunnable;
+import org.assertj.core.api.InstanceOfAssertFactories;
+import org.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.RegisterExtension;
import java.io.IOException;
import java.net.ServerSocket;
@@ -56,26 +54,24 @@ import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
-import static org.hamcrest.Matchers.containsString;
-import static org.hamcrest.Matchers.empty;
-import static org.hamcrest.Matchers.equalTo;
-import static org.hamcrest.Matchers.hasSize;
-import static org.hamcrest.Matchers.instanceOf;
-import static org.junit.Assert.assertThat;
-import static org.junit.Assert.assertThrows;
+import static org.assertj.core.api.Assertions.as;
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.assertj.core.api.Assertions.assertThatThrownBy;
+import static org.assertj.core.api.InstanceOfAssertFactories.THROWABLE;
/** Tests for {@link RestClient}. */
-public class RestClientTest extends TestLogger {
- @ClassRule
- public static final TestExecutorResource<ScheduledExecutorService> EXECUTOR_RESOURCE =
- TestingUtils.defaultExecutorResource();
+class RestClientTest {
+
+ @RegisterExtension
+ static final TestExecutorExtension<ScheduledExecutorService> EXECUTOR_EXTENSION =
+ TestingUtils.defaultExecutorExtension();
private static final String unroutableIp = "240.0.0.0";
private static final long TIMEOUT = 10L;
@Test
- public void testConnectionTimeout() throws Exception {
+ void testConnectionTimeout() throws Exception {
final Configuration config = new Configuration();
config.setLong(RestOptions.CONNECTION_TIMEOUT, 1);
try (final RestClient restClient = new RestClient(config, Executors.directExecutor())) {
@@ -87,9 +83,11 @@ public class RestClientTest extends TestLogger {
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance());
- final Throwable cause = assertThrows(ExecutionException.class, future::get).getCause();
- assertThat(cause, instanceOf(ConnectTimeoutException.class));
- assertThat(cause.getMessage(), containsString(unroutableIp));
+ FlinkAssertions.assertThatFuture(future)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(ConnectTimeoutException.class)
+ .extracting(Throwable::getCause, as(InstanceOfAssertFactories.THROWABLE))
+ .hasMessageContaining(unroutableIp);
}
}
@@ -97,18 +95,18 @@ public class RestClientTest extends TestLogger {
public void testInvalidVersionRejection() throws Exception {
try (final RestClient restClient =
new RestClient(new Configuration(), Executors.directExecutor())) {
- CompletableFuture<EmptyResponseBody> invalidVersionResponse =
- restClient.sendRequest(
- unroutableIp,
- 80,
- new TestMessageHeaders(),
- EmptyMessageParameters.getInstance(),
- EmptyRequestBody.getInstance(),
- Collections.emptyList(),
- RuntimeRestAPIVersion.V0);
- Assert.fail("The request should have been rejected due to a version mismatch.");
- } catch (IllegalArgumentException e) {
- // expected
+ assertThatThrownBy(
+ () ->
+ restClient.sendRequest(
+ unroutableIp,
+ 80,
+ new TestMessageHeaders(),
+ EmptyMessageParameters.getInstance(),
+ EmptyRequestBody.getInstance(),
+ Collections.emptyList(),
+ RuntimeRestAPIVersion.V0))
+ .as("The request should have been rejected due to a version mismatch.")
+ .isInstanceOf(IllegalArgumentException.class);
}
}
@@ -119,7 +117,7 @@ public class RestClientTest extends TestLogger {
config.setLong(RestOptions.IDLENESS_TIMEOUT, 5000L);
try (final ServerSocket serverSocket = new ServerSocket(0);
final RestClient restClient =
- new RestClient(config, EXECUTOR_RESOURCE.getExecutor())) {
+ new RestClient(config, EXECUTOR_EXTENSION.getExecutor())) {
final String targetAddress = "localhost";
final int targetPort = serverSocket.getLocalPort();
@@ -153,13 +151,9 @@ public class RestClientTest extends TestLogger {
connectionSocket.close();
}
- try {
- responseFuture.get();
- } catch (ExecutionException ee) {
- if (!ExceptionUtils.findThrowable(ee, IOException.class).isPresent()) {
- throw ee;
- }
- }
+ FlinkAssertions.assertThatFuture(responseFuture)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IOException.class);
}
}
@@ -173,7 +167,7 @@ public class RestClientTest extends TestLogger {
try (final ServerSocket serverSocket = new ServerSocket(0);
final RestClient restClient =
- new RestClient(config, EXECUTOR_RESOURCE.getExecutor())) {
+ new RestClient(config, EXECUTOR_EXTENSION.getExecutor())) {
final String targetAddress = "localhost";
final int targetPort = serverSocket.getLocalPort();
@@ -202,13 +196,9 @@ public class RestClientTest extends TestLogger {
restClient.close();
- try {
- responseFuture.get();
- } catch (ExecutionException ee) {
- if (!ExceptionUtils.findThrowable(ee, IOException.class).isPresent()) {
- throw ee;
- }
- }
+ FlinkAssertions.assertThatFuture(responseFuture)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IOException.class);
} finally {
if (connectionSocket != null) {
connectionSocket.close();
@@ -236,14 +226,11 @@ public class RestClientTest extends TestLogger {
EmptyMessageParameters.getInstance(),
EmptyRequestBody.getInstance());
- // Call get() on the future with a timeout of 0s so we can test that the exception
- // thrown is not a TimeoutException, which is what would be thrown if restClient were
- // not already closed
- final ThrowingRunnable getFuture = () -> future.get(0, TimeUnit.SECONDS);
-
- final Throwable cause = assertThrows(ExecutionException.class, getFuture).getCause();
- assertThat(cause, instanceOf(IllegalStateException.class));
- assertThat(cause.getMessage(), equalTo("RestClient is already closed"));
+ FlinkAssertions.assertThatFuture(future)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IllegalStateException.class)
+ .extracting(Throwable::getCause, as(THROWABLE))
+ .hasMessage("RestClient is already closed");
}
}
@@ -270,7 +257,7 @@ public class RestClientTest extends TestLogger {
new Configuration(), Executors.directExecutor(), selectStrategyFactory)) {
// Check that client's internal collection of pending response futures is empty prior to
// the request
- assertThat(restClient.getResponseChannelFutures(), empty());
+ assertThat(restClient.getResponseChannelFutures()).isEmpty();
final CompletableFuture<?> requestFuture =
restClient.sendRequest(
@@ -282,7 +269,7 @@ public class RestClientTest extends TestLogger {
// Check that client's internal collection of pending response futures now has one
// entry, presumably due to the call to sendRequest
- assertThat(restClient.getResponseChannelFutures(), hasSize(1));
+ assertThat(restClient.getResponseChannelFutures()).hasSize(1);
// Wait for Netty to start connecting, then while it's paused in the SelectStrategy,
// close the client before unpausing Netty
@@ -290,16 +277,15 @@ public class RestClientTest extends TestLogger {
final CompletableFuture<Void> closeFuture = restClient.closeAsync();
closeTriggered.trigger();
- // Close should complete successfully
- closeFuture.get();
+ FlinkAssertions.assertThatFuture(closeFuture)
+ .as("Close should have had completed.")
+ .eventuallySucceeds();
- final Throwable cause =
- assertThrows(
- ExecutionException.class,
- () -> requestFuture.get(0, TimeUnit.SECONDS))
- .getCause();
- assertThat(cause, instanceOf(IllegalStateException.class));
- assertThat(cause.getMessage(), equalTo("executor not accepting a task"));
+ FlinkAssertions.assertThatFuture(requestFuture)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IllegalStateException.class)
+ .extracting(Throwable::getCause, as(THROWABLE))
+ .hasMessage("executor not accepting a task");
}
}
@@ -318,15 +304,13 @@ public class RestClientTest extends TestLogger {
// Ensure the client's internal collection of pending response futures was cleared after
// close
- assertThat(restClient.getResponseChannelFutures(), empty());
-
- final Throwable cause =
- assertThrows(
- ExecutionException.class,
- () -> responseChannelFuture.get(0, TimeUnit.SECONDS))
- .getCause();
- assertThat(cause, instanceOf(IllegalStateException.class));
- assertThat(cause.getMessage(), equalTo("RestClient closed before request completed"));
+ assertThat(restClient.getResponseChannelFutures()).isEmpty();
+
+ FlinkAssertions.assertThatFuture(responseChannelFuture)
+ .eventuallyFailsWith(ExecutionException.class)
+ .withCauseInstanceOf(IllegalStateException.class)
+ .extracting(Throwable::getCause, as(THROWABLE))
+ .hasMessage("RestClient closed before request completed");
}
}
[flink] 02/02: [FLINK-32751][runtime] Refactors CollectSinkOperatorCoordinator to enable case-specific logging
Posted by ma...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
mapohl pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit b5b1340f91d6cd2deb563eb67d8cbb3d1852544d
Author: Matthias Pohl <ma...@aiven.io>
AuthorDate: Wed Aug 9 08:58:04 2023 +0200
[FLINK-32751][runtime] Refactors CollectSinkOperatorCoordinator to enable case-specific logging
shutdownNow() is added to immediately stop any already submitted requests. That makes
the close logic more explicit rather than relying on the Runnables to finish implicitly
through the connection loss.
Any ongoing requests are now also cancelled. This allows us to log specific cases of unexpected
errors properly.
---
.../collect/CollectSinkOperatorCoordinator.java | 140 ++++++++++++---------
1 file changed, 82 insertions(+), 58 deletions(-)
diff --git a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorCoordinator.java b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorCoordinator.java
index e926d625e0a..806ed180496 100644
--- a/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorCoordinator.java
+++ b/flink-streaming-java/src/main/java/org/apache/flink/streaming/api/operators/collect/CollectSinkOperatorCoordinator.java
@@ -25,8 +25,10 @@ import org.apache.flink.runtime.operators.coordination.CoordinationRequestHandle
import org.apache.flink.runtime.operators.coordination.CoordinationResponse;
import org.apache.flink.runtime.operators.coordination.OperatorCoordinator;
import org.apache.flink.runtime.operators.coordination.OperatorEvent;
+import org.apache.flink.util.ExceptionUtils;
import org.apache.flink.util.Preconditions;
import org.apache.flink.util.concurrent.ExecutorThreadFactory;
+import org.apache.flink.util.concurrent.FutureUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
@@ -41,7 +43,10 @@ import java.io.ObjectOutputStream;
import java.net.InetSocketAddress;
import java.net.Socket;
import java.util.Collections;
+import java.util.Set;
+import java.util.concurrent.CancellationException;
import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
@@ -63,6 +68,9 @@ public class CollectSinkOperatorCoordinator
private DataInputViewStreamWrapper inStream;
private DataOutputViewStreamWrapper outStream;
+ private final Set<CompletableFuture<CoordinationResponse>> ongoingRequests =
+ ConcurrentHashMap.newKeySet();
+
private ExecutorService executorService;
public CollectSinkOperatorCoordinator(int socketTimeout) {
@@ -79,8 +87,9 @@ public class CollectSinkOperatorCoordinator
@Override
public void close() throws Exception {
+ LOG.info("Closing the CollectSinkOperatorCoordinator.");
+ this.executorService.shutdownNow();
closeConnection();
- this.executorService.shutdown();
}
@Override
@@ -101,75 +110,82 @@ public class CollectSinkOperatorCoordinator
"Coordination request must be a CollectCoordinationRequest");
CollectCoordinationRequest collectRequest = (CollectCoordinationRequest) request;
- CompletableFuture<CoordinationResponse> responseFuture = new CompletableFuture<>();
-
if (address == null) {
- completeWithEmptyResponse(collectRequest, responseFuture);
- return responseFuture;
+ return CompletableFuture.completedFuture(createEmptyResponse(collectRequest));
}
- executorService.submit(() -> handleRequestImpl(collectRequest, responseFuture, address));
- return responseFuture;
+ final CompletableFuture<CoordinationResponse> responseFuture =
+ FutureUtils.supplyAsync(
+ () -> handleRequestImpl(collectRequest, address), executorService);
+
+ ongoingRequests.add(responseFuture);
+ return responseFuture.handle(
+ (response, error) -> {
+ ongoingRequests.remove(responseFuture);
+
+ if (response != null) {
+ return response;
+ }
+
+ // cancelling the future implies that the error handling happens somewhere else
+ if (!ExceptionUtils.findThrowable(error, CancellationException.class)
+ .isPresent()) {
+ // Request failed: Close current connection and send back empty results
+ // we catch every exception here because the Socket might suddenly become
+ // null. We don't want the coordinator to fail if the sink fails.
+ if (LOG.isDebugEnabled()) {
+ LOG.warn(
+ "Collect sink coordinator encountered an unexpected error.",
+ error);
+ } else {
+ LOG.warn(
+ "Collect sink coordinator encounters a {}: {}",
+ error.getClass().getSimpleName(),
+ error.getMessage());
+ }
+
+ closeConnection();
+ }
+
+ return createEmptyResponse(collectRequest);
+ });
}
- private void handleRequestImpl(
- CollectCoordinationRequest request,
- CompletableFuture<CoordinationResponse> responseFuture,
- InetSocketAddress sinkAddress) {
+ private CoordinationResponse handleRequestImpl(
+ CollectCoordinationRequest request, InetSocketAddress sinkAddress) throws IOException {
if (sinkAddress == null) {
- closeConnection();
- completeWithEmptyResponse(request, responseFuture);
- return;
+ throw new NullPointerException("No sinkAddress available.");
}
- try {
- if (socket == null) {
- socket = new Socket();
- socket.setSoTimeout(socketTimeout);
- socket.setKeepAlive(true);
- socket.setTcpNoDelay(true);
-
- socket.connect(sinkAddress);
- inStream = new DataInputViewStreamWrapper(socket.getInputStream());
- outStream = new DataOutputViewStreamWrapper(socket.getOutputStream());
- LOG.info("Sink connection established");
- }
+ if (socket == null) {
+ socket = new Socket();
+ socket.setSoTimeout(socketTimeout);
+ socket.setKeepAlive(true);
+ socket.setTcpNoDelay(true);
- // send version and offset to sink server
- if (LOG.isDebugEnabled()) {
- LOG.debug("Forwarding request to sink socket server");
- }
- request.serialize(outStream);
-
- // fetch back serialized results
- if (LOG.isDebugEnabled()) {
- LOG.debug("Fetching serialized result from sink socket server");
- }
- responseFuture.complete(new CollectCoordinationResponse(inStream));
- } catch (Exception e) {
- // request failed, close current connection and send back empty results
- // we catch every exception here because socket might suddenly becomes null if the sink
- // fails
- // and we do not want the coordinator to fail
- if (LOG.isDebugEnabled()) {
- // this is normal when sink restarts or job ends, so we print a debug log
- LOG.debug("Collect sink coordinator encounters an exception", e);
- }
- closeConnection();
- completeWithEmptyResponse(request, responseFuture);
+ socket.connect(sinkAddress);
+ inStream = new DataInputViewStreamWrapper(socket.getInputStream());
+ outStream = new DataOutputViewStreamWrapper(socket.getOutputStream());
+ LOG.info("Sink connection established");
}
+
+ // send version and offset to sink server
+ LOG.debug("Forwarding request to sink socket server");
+ request.serialize(outStream);
+
+ // fetch back serialized results
+ LOG.debug("Fetching serialized result from sink socket server");
+ return new CollectCoordinationResponse(inStream);
}
- private void completeWithEmptyResponse(
- CollectCoordinationRequest request, CompletableFuture<CoordinationResponse> future) {
- future.complete(
- new CollectCoordinationResponse(
- request.getVersion(),
- // this lastCheckpointedOffset is OK
- // because client will only expose results to the users when the
- // checkpointed offset increases
- -1,
- Collections.emptyList()));
+ private CollectCoordinationResponse createEmptyResponse(CollectCoordinationRequest request) {
+ return new CollectCoordinationResponse(
+ request.getVersion(),
+ // this lastCheckpointedOffset is OK
+ // because client will only expose results to the users when the
+ // checkpointed offset increases
+ -1,
+ Collections.emptyList());
}
private void closeConnection() {
@@ -217,6 +233,9 @@ public class CollectSinkOperatorCoordinator
throws Exception {
if (checkpointData == null) {
// restore before any checkpoint completed
+ LOG.info("Any ongoing requests are cancelled due to a coordinator reset.");
+ cancelOngoingRequests();
+
closeConnection();
} else {
ByteArrayInputStream bais = new ByteArrayInputStream(checkpointData);
@@ -225,6 +244,11 @@ public class CollectSinkOperatorCoordinator
}
}
+ private void cancelOngoingRequests() {
+ ongoingRequests.forEach(ft -> ft.cancel(true));
+ ongoingRequests.clear();
+ }
+
/** Provider for {@link CollectSinkOperatorCoordinator}. */
public static class Provider implements OperatorCoordinator.Provider {