You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by pn...@apache.org on 2023/01/27 09:15:04 UTC
[flink] 03/03: [hotfix][runtime] Check style ServerTransportErrorHandlingTest and ClientTransportErrorHandlingTest fix
This is an automated email from the ASF dual-hosted git repository.
pnowojski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/flink.git
commit 952c7831e7411201edea7fdf205b481c0ae26a8e
Author: Anton Kalashnikov <ak...@gmail.com>
AuthorDate: Thu Jan 26 17:56:30 2023 +0100
[hotfix][runtime] Check style ServerTransportErrorHandlingTest and ClientTransportErrorHandlingTest fix
---
.../netty/ClientTransportErrorHandlingTest.java | 79 ++++++++++------------
.../netty/ServerTransportErrorHandlingTest.java | 30 ++++----
2 files changed, 47 insertions(+), 62 deletions(-)
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ClientTransportErrorHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ClientTransportErrorHandlingTest.java
index c8ae23cac08..47d312f5ffa 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ClientTransportErrorHandlingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ClientTransportErrorHandlingTest.java
@@ -42,7 +42,6 @@ import org.apache.flink.shaded.netty4.io.netty.channel.ChannelPromise;
import org.apache.flink.shaded.netty4.io.netty.channel.embedded.EmbeddedChannel;
import org.junit.jupiter.api.Test;
-import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import java.io.IOException;
@@ -55,8 +54,8 @@ import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.initServer
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.shutdown;
import static org.assertj.core.api.Assertions.assertThat;
import static org.assertj.core.api.Assertions.assertThatNoException;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.isA;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.isA;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.doThrow;
import static org.mockito.Mockito.mock;
@@ -102,9 +101,7 @@ class ClientTransportErrorHandlingTest {
@Override
public void write(
- ChannelHandlerContext ctx, Object msg, ChannelPromise promise)
- throws Exception {
-
+ ChannelHandlerContext ctx, Object msg, ChannelPromise promise) {
if (writeNum >= 1) {
throw new RuntimeException("Expected test exception.");
}
@@ -124,17 +121,15 @@ class ClientTransportErrorHandlingTest {
final CountDownLatch sync = new CountDownLatch(1);
- // Do this with explicit synchronization. Otherwise this is not robust against slow timings
+ // Do this with explicit synchronization. Otherwise, this is not robust against slow timings
// of the callback (e.g. we cannot just verify that it was called once, because there is
// a chance that we do this too early).
doAnswer(
- new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- sync.countDown();
- return null;
- }
- })
+ (Answer<Void>)
+ invocation -> {
+ sync.countDown();
+ return null;
+ })
.when(rich[1])
.onError(isA(LocalTransportException.class));
@@ -143,7 +138,7 @@ class ClientTransportErrorHandlingTest {
// Second request is *not* successful
requestClient.requestSubpartition(new ResultPartitionID(), 0, rich[1], 0);
- // Wait for the notification and it could confirm all the request operations are done
+ // Wait for the notification, and it could confirm all the request operations are done
assertThat(sync.await(TestingUtils.TESTING_DURATION.toMillis(), TimeUnit.MILLISECONDS))
.withFailMessage(
"Timed out after waiting for "
@@ -227,9 +222,7 @@ class ClientTransportErrorHandlingTest {
// Close on read
new ChannelInboundHandlerAdapter() {
@Override
- public void channelRead(ChannelHandlerContext ctx, Object msg)
- throws Exception {
-
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.channel().close();
}
}
@@ -250,12 +243,9 @@ class ClientTransportErrorHandlingTest {
final CountDownLatch sync = new CountDownLatch(rich.length);
Answer<Void> countDownLatch =
- new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- sync.countDown();
- return null;
- }
+ invocation -> {
+ sync.countDown();
+ return null;
};
for (RemoteInputChannel r : rich) {
@@ -329,26 +319,27 @@ class ClientTransportErrorHandlingTest {
// Verify the Exception
doAnswer(
- new Answer<Void>() {
- @Override
- public Void answer(InvocationOnMock invocation) throws Throwable {
- Throwable cause = (Throwable) invocation.getArguments()[0];
-
- try {
- assertThat(cause).isInstanceOf(RemoteTransportException.class);
- assertThat(cause)
- .hasMessageNotContaining("Connection reset by peer");
-
- assertThat(cause.getCause()).isInstanceOf(IOException.class);
- assertThat(cause.getCause())
- .hasMessage("Connection reset by peer");
- } catch (Throwable t) {
- error[0] = t;
- }
-
- return null;
- }
- })
+ (Answer<Void>)
+ invocation -> {
+ Throwable cause = (Throwable) invocation.getArguments()[0];
+
+ try {
+ assertThat(cause)
+ .isInstanceOf(RemoteTransportException.class);
+ assertThat(cause)
+ .hasMessageNotContaining(
+ "Connection reset by peer");
+
+ assertThat(cause.getCause())
+ .isInstanceOf(IOException.class);
+ assertThat(cause.getCause())
+ .hasMessage("Connection reset by peer");
+ } catch (Throwable t) {
+ error[0] = t;
+ }
+
+ return null;
+ })
.when(rich)
.onError(any(Throwable.class));
diff --git a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
index 55985168ac7..c8faf422fc0 100644
--- a/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
+++ b/flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/ServerTransportErrorHandlingTest.java
@@ -19,6 +19,7 @@
package org.apache.flink.runtime.io.network.netty;
import org.apache.flink.runtime.io.network.TaskEventDispatcher;
+import org.apache.flink.runtime.io.network.netty.CancelPartitionRequestTest.InfiniteSubpartitionView;
import org.apache.flink.runtime.io.network.partition.BufferAvailabilityListener;
import org.apache.flink.runtime.io.network.partition.ResultPartitionID;
import org.apache.flink.runtime.io.network.partition.ResultPartitionManager;
@@ -33,7 +34,6 @@ import org.apache.flink.shaded.netty4.io.netty.channel.ChannelHandlerContext;
import org.apache.flink.shaded.netty4.io.netty.channel.ChannelInboundHandlerAdapter;
import org.junit.Test;
-import org.mockito.invocation.InvocationOnMock;
import org.mockito.stubbing.Answer;
import java.util.concurrent.CountDownLatch;
@@ -43,8 +43,8 @@ import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.connect;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.initServerAndClient;
import static org.apache.flink.runtime.io.network.netty.NettyTestUtil.shutdown;
import static org.junit.Assert.fail;
-import static org.mockito.Matchers.any;
-import static org.mockito.Matchers.anyInt;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.ArgumentMatchers.anyInt;
import static org.mockito.Mockito.mock;
import static org.mockito.Mockito.when;
@@ -64,18 +64,14 @@ public class ServerTransportErrorHandlingTest {
anyInt(),
any(BufferAvailabilityListener.class)))
.thenAnswer(
- new Answer<ResultSubpartitionView>() {
- @Override
- public ResultSubpartitionView answer(InvocationOnMock invocationOnMock)
- throws Throwable {
- BufferAvailabilityListener listener =
- (BufferAvailabilityListener)
- invocationOnMock.getArguments()[2];
- listener.notifyDataAvailable();
- return new CancelPartitionRequestTest.InfiniteSubpartitionView(
- outboundBuffers, sync);
- }
- });
+ (Answer<ResultSubpartitionView>)
+ invocationOnMock -> {
+ BufferAvailabilityListener listener =
+ (BufferAvailabilityListener)
+ invocationOnMock.getArguments()[2];
+ listener.notifyDataAvailable();
+ return new InfiniteSubpartitionView(outboundBuffers, sync);
+ });
NettyProtocol protocol =
new NettyProtocol(partitionManager, mock(TaskEventDispatcher.class)) {
@@ -87,9 +83,7 @@ public class ServerTransportErrorHandlingTest {
// Close on read
new ChannelInboundHandlerAdapter() {
@Override
- public void channelRead(ChannelHandlerContext ctx, Object msg)
- throws Exception {
-
+ public void channelRead(ChannelHandlerContext ctx, Object msg) {
ctx.channel().close();
}
}