flink-instrumentation/workspace/flink-1.4.1-instrumented/flink-1.4.1/flink-queryable-state/flink-queryable-state-runtime/src/test/java/org/apache/flink/queryablestate/network/ClientTest.java
/**
	 * Tests that a request to an unavailable host is failed with ConnectException.
	 */
Tests that a request to an unavailable host is failed with ConnectException.
Tests that a request to an unavailable host is failed with ConnectException .
@Test
	public void testRequestUnavailableHost() throws Exception {
		Deadline deadline = TEST_TIMEOUT.fromNow();
		AtomicKvStateRequestStats stats = new AtomicKvStateRequestStats();
		MessageSerializer<KvStateInternalRequest, KvStateResponse> serializer =
				new MessageSerializer<>(new KvStateInternalRequest.KvStateInternalRequestDeserializer(), new KvStateResponse.KvStateResponseDeserializer());
		Client<KvStateInternalRequest, KvStateResponse> client = null;
		try {
			client = new Client<>("Test Client", 1, serializer, stats);
			int availablePort = NetUtils.getAvailablePort();
			InetSocketAddress serverAddress = new InetSocketAddress(
					InetAddress.getLocalHost(),
					availablePort);
			KvStateInternalRequest request = new KvStateInternalRequest(new KvStateID(), new byte[0]);
			CompletableFuture<KvStateResponse> future = client.sendRequest(serverAddress, request);
			try {
				future.get(deadline.timeLeft().toMillis(), TimeUnit.MILLISECONDS);
				fail("Did not throw expected ConnectException");
			} catch (ExecutionException e) {
				if (!(e.getCause() instanceof ConnectException)) {
					fail("Did not throw expected ConnectException");
				}
			}
		} finally {
			if (client != null) {
				try {
					client.shutdown().get(10L, TimeUnit.SECONDS);
				} catch (Exception e) {
					e.printStackTrace();
				}
				Assert.assertTrue(client.isEventGroupShutdown());
			}
			assertEquals("Channel leak", 0L, stats.getNumConnections());
		}
	}
@ Test public void testRequestUnavailableHost ( ) throws Exception { Deadline deadline = TEST_TIMEOUT . fromNow ( ) ; AtomicKvStateRequestStats stats = new AtomicKvStateRequestStats ( ) ; MessageSerializer < KvStateInternalRequest , KvStateResponse > serializer = new MessageSerializer < > ( new KvStateInternalRequest . KvStateInternalRequestDeserializer ( ) , new KvStateResponse . KvStateResponseDeserializer ( ) ) ; Client < KvStateInternalRequest , KvStateResponse > client = null ; try { client = new Client < > ( "Test Client" , 1 , serializer , stats ) ; int availablePort = NetUtils . getAvailablePort ( ) ; InetSocketAddress serverAddress = new InetSocketAddress ( InetAddress . getLocalHost ( ) , availablePort ) ; KvStateInternalRequest request = new KvStateInternalRequest ( new KvStateID ( ) , new byte [ 0 ] ) ; CompletableFuture < KvStateResponse > future = client . sendRequest ( serverAddress , request ) ; try { future . get ( deadline . timeLeft ( ) . toMillis ( ) , TimeUnit . MILLISECONDS ) ; fail ( "Did not throw expected ConnectException" ) ; } catch ( ExecutionException e ) { if ( ! ( e . getCause ( ) instanceof ConnectException ) ) { fail ( "Did not throw expected ConnectException" ) ; } } } finally { if ( client != null ) { try { client . shutdown ( ) . get ( 10L , TimeUnit . SECONDS ) ; } catch ( Exception e ) { e . printStackTrace ( ) ; } Assert . assertTrue ( client . isEventGroupShutdown ( ) ) ; } assertEquals ( "Channel leak" , 0L , stats . getNumConnections ( ) ) ; } }
Tests that a request to an unavailable host is failed with ConnectException.
Tests that a request to an unavailable host is failed with ConnectException .
// else expected