[refactor][tests][network] Cleanup PartitionRequestClientFactoryTest

flink-runtime/src/test/java/org/apache/flink/runtime/io/network/netty/PartitionRequestClientFactoryTest.java

@@ -38,10 +38,8 @@
 import java.net.InetAddress;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
-import java.net.UnknownHostException;
 import java.util.ArrayList;
 import java.util.List;
-import java.util.concurrent.Callable;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
@@ -131,8 +129,6 @@ public void testNettyClientConnectRetryFailure() throws Exception {
 
  factory.createPartitionRequestClient(serverAndClient.getConnectionID(0));
 
- } catch (Exception e) {
- throw e;
  } finally {
  serverAndClient.client().shutdown();
  serverAndClient.server().shutdown();
@@ -150,24 +146,21 @@ public void testNettyClientConnectRetryMultipleThread() throws Exception {
  List<Future<NettyPartitionRequestClient>> futures = new ArrayList<>();
 
  for (int i = 0; i < 10; i++) {
- Future<NettyPartitionRequestClient> future = threadPoolExecutor.submit(new Callable<NettyPartitionRequestClient>() {
- @Override
- public NettyPartitionRequestClient call() {
- NettyPartitionRequestClient client = null;
- try {
- client = factory.createPartitionRequestClient(serverAndClient.getConnectionID(0));
- } catch (Exception e) {
- fail(e.getMessage());
- }
- return client;
+ Future<NettyPartitionRequestClient> future = threadPoolExecutor.submit(() -> {
+ NettyPartitionRequestClient client = null;
+ try {
+ client = factory.createPartitionRequestClient(serverAndClient.getConnectionID(0));
+ } catch (Exception e) {
+ fail(e.getMessage());
  }
+ return client;
  });
 
  futures.add(future);
  }
 
  futures.forEach(runnableFuture -> {
- NettyPartitionRequestClient client = null;
+ NettyPartitionRequestClient client;
  try {
  client = runnableFuture.get();
  assertNotNull(client);
@@ -183,7 +176,7 @@ public NettyPartitionRequestClient call() {
  }
 
  private NettyTestUtil.NettyServerAndClient createNettyServerAndClient() throws Exception {
- NettyTestUtil.NettyServerAndClient serverAndClient = NettyTestUtil.initServerAndClient(
+ return NettyTestUtil.initServerAndClient(
  new NettyProtocol(null, null) {
 
  @Override
@@ -196,17 +189,15 @@ public ChannelHandler[] getClientChannelHandlers () {
  return new ChannelHandler[]{mock(NetworkClientHandler.class)};
  }
  });
-
- return serverAndClient;
  }
 
  private static class UnstableNettyClient extends NettyClient {
 
- private NettyClient nettyClient;
+ private final NettyClient nettyClient;
 
  private int retry;
 
- public UnstableNettyClient(NettyClient nettyClient, int retry) {
+ UnstableNettyClient(NettyClient nettyClient, int retry) {
  super(null);
  this.nettyClient = nettyClient;
  this.retry = retry;
@@ -225,7 +216,7 @@ ChannelFuture connect(final InetSocketAddress serverSocketAddress) {
 
  private static class FailingNettyClient extends NettyClient {
 
- public FailingNettyClient() {
+ FailingNettyClient() {
  super(null);
  }
 
@@ -235,7 +226,7 @@ ChannelFuture connect(final InetSocketAddress serverSocketAddress) {
  }
  }
 
- private class AwaitingNettyClient extends NettyClient {
+ private static class AwaitingNettyClient extends NettyClient {
  private volatile boolean awaitForInterrupts;
  private final NettyClient client;
 
@@ -266,14 +257,14 @@ public CountDownLatchOnConnectHandler(CountDownLatch syncOnConnect) {
  }
 
  @Override
- public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) throws Exception {
+ public void connect(ChannelHandlerContext ctx, SocketAddress remoteAddress, SocketAddress localAddress, ChannelPromise promise) {
  syncOnConnect.countDown();
  }
  }
 
  private static class UncaughtTestExceptionHandler implements UncaughtExceptionHandler {
 
- private final List<Throwable> errors = new ArrayList<Throwable>(1);
+ private final List<Throwable> errors = new ArrayList<>(1);
 
  @Override
  public void uncaughtException(Thread t, Throwable e) {
@@ -310,10 +301,6 @@ private static Tuple2<NettyServer, NettyClient> createNettyServerAndClient(Netty
  }
  }
 
- return new Tuple2<NettyServer, NettyClient>(server, client);
- }
-
- private static ConnectionID createServerConnectionID(int connectionIndex) throws UnknownHostException {
- return new ConnectionID(new InetSocketAddress(InetAddress.getLocalHost(), SERVER_PORT), connectionIndex);
+ return new Tuple2<>(server, client);
  }
 }