java.nio.channels.CancelledKeyException#java.net.ConnectException源码实例Demo

@Override
public Observable<?> call(Observable<? extends Throwable> observable) {
    return observable
            .zipWith(Observable.range(1, count + 1), new Func2<Throwable, Integer, Wrapper>() {
                @Override
                public Wrapper call(Throwable throwable, Integer integer) {
                    return new Wrapper(throwable, integer);
                }
            }).flatMap(new Func1<Wrapper, Observable<?>>() {
                @Override
                public Observable<?> call(Wrapper wrapper) {
                    if ((wrapper.throwable instanceof ConnectException
                            || wrapper.throwable instanceof SocketTimeoutException
                            || wrapper.throwable instanceof TimeoutException)
                            && wrapper.index < count + 1) { //如果超出重试次数也抛出错误，否则默认是会进入onCompleted
                        return Observable.timer(delay + (wrapper.index - 1) * increaseDelay, TimeUnit.MILLISECONDS);

                    }
                    return Observable.error(wrapper.throwable);
                }
            });
}
 

/**
 * Test that we can't accidentally connect back to the connecting socket due
 * to a quirk in the TCP spec.
 *
 * This is a regression test for HADOOP-6722.
 */
@Test
public void testAvoidLoopbackTcpSockets() throws Exception {
  Configuration conf = new Configuration();

  Socket socket = NetUtils.getDefaultSocketFactory(conf)
    .createSocket();
  socket.bind(new InetSocketAddress("127.0.0.1", 0));
  System.err.println("local address: " + socket.getLocalAddress());
  System.err.println("local port: " + socket.getLocalPort());
  try {
    NetUtils.connect(socket,
      new InetSocketAddress(socket.getLocalAddress(), socket.getLocalPort()),
      20000);
    socket.close();
    fail("Should not have connected");
  } catch (ConnectException ce) {
    System.err.println("Got exception: " + ce);
    assertTrue(ce.getMessage().contains("resulted in a loopback"));
  } catch (SocketException se) {
    // Some TCP stacks will actually throw their own Invalid argument exception
    // here. This is also OK.
    assertTrue(se.getMessage().contains("Invalid argument"));
  }
}
 

private void connect(Collection<DataTreeModification<Node>> changes) {
    for (DataTreeModification<Node> change : changes) {
        final InstanceIdentifier<Node> key = change.getRootPath().getRootIdentifier();
        final DataObjectModification<Node> mod = change.getRootNode();
        Node node = getCreated(mod);
        if (node != null) {
            HwvtepGlobalAugmentation hwvtepGlobal = node.augmentation(HwvtepGlobalAugmentation.class);
            // We can only connect if user configured connection info
            if (hwvtepGlobal != null && hwvtepGlobal.getConnectionInfo() != null) {
                ConnectionInfo connection = hwvtepGlobal.getConnectionInfo();
                InstanceIdentifier<Node> iid = hcm.getInstanceIdentifier(connection);
                if (iid != null) {
                    LOG.warn("Connection to device {} already exists. Plugin does not allow multiple connections "
                                    + "to same device, hence dropping the request {}", connection, hwvtepGlobal);
                } else {
                    try {
                        hcm.connect(key, hwvtepGlobal);
                    } catch (UnknownHostException | ConnectException e) {
                        LOG.warn("Failed to connect to HWVTEP node", e);
                    }
                }
            }
        }
    }
}
 

public PTZStatus getStatus(String profileToken) {
	GetStatus request = new GetStatus();
	GetStatusResponse response = new GetStatusResponse();

	request.setProfileToken(profileToken);

	try {
		response = (GetStatusResponse) soap.createSOAPPtzRequest(request, response, false);
	}
	catch (SOAPException | ConnectException e) {
		e.printStackTrace();
		return null;
	}

	if (response == null) {
		return null;
	}

	return response.getPTZStatus();
}
 

@Override
public void onThrowable(Throwable t) {
    inflightRequests.remove(this);
    job.setState("HTTP.onThrowable");
    MetricUtils.httpRequestLatencyMetric(job, TimeUtils.getElapseTime(startTime));
    MetricUtils.httpRequestFailureMetric(job, null);
    String errorLog = String.format("Action Result: HttpAccess[result:exception,url:%s,request:%s,used:%d],Exception:%s|%s",
            getUrl(), this, TimeUtils.getElapseTime(startTime), t.getClass().getSimpleName(), t.getMessage());
    if (t instanceof ConnectException || t instanceof TimeoutException) {
        LOGGER.info(errorLog);
    } else {
        LogUtils.logErrorInfo("CarreraAsyncRequest_error", errorLog, t);
    }
    delayRetryRequest(DEFAULT_RETRY_DELAY << Math.min(job.getErrorRetryCnt(), MAX_RETRY_DELAY_FACTOR));
    job.incErrorRetryCnt();
}
 

protected <A> Answer getPAMInstanceAnswer(A validAnswer) {
    return invocation -> {
        KieBpmConfig config = invocation.getArgument(0);
        switch (config.getName()) {
            case "default":
                return Arrays.asList(validAnswer);
            case "broken":
                throw new ApsSystemException("",
                        new HttpServerErrorException(HttpStatus.INTERNAL_SERVER_ERROR));
            case "unreachable":
                throw new ApsSystemException("", new ConnectException());
            default:
                return null;
        }
    };
}
 

private static ClientResponse mockEntityClientResponse(
    TimelineClientImpl client, ClientResponse.Status status,
    boolean hasError, boolean hasRuntimeError) {
  ClientResponse response = mock(ClientResponse.class);
  if (hasRuntimeError) {
    doThrow(new ClientHandlerException(new ConnectException())).when(client)
        .doPostingObject(any(TimelineEntities.class), any(String.class));
    return response;
  }
  doReturn(response).when(client)
      .doPostingObject(any(TimelineEntities.class), any(String.class));
  when(response.getClientResponseStatus()).thenReturn(status);
  TimelinePutResponse.TimelinePutError error =
      new TimelinePutResponse.TimelinePutError();
  error.setEntityId("test entity id");
  error.setEntityType("test entity type");
  error.setErrorCode(TimelinePutResponse.TimelinePutError.IO_EXCEPTION);
  TimelinePutResponse putResponse = new TimelinePutResponse();
  if (hasError) {
    putResponse.addError(error);
  }
  when(response.getEntity(TimelinePutResponse.class)).thenReturn(putResponse);
  return response;
}
 

@Override
protected void connectOP(SelectionKey key) throws IOException {
	U.must(key.isConnectable());

	SocketChannel socketChannel = (SocketChannel) key.channel();
	if (!socketChannel.isConnectionPending()) {
		// not ready to retrieve the connection status
		return;
	}

	ConnectionTarget target = (ConnectionTarget) key.attachment();

	boolean ready;
	try {
		ready = socketChannel.finishConnect();
		U.must(ready, "Expected an established connection!");

		Log.info("Connected", "address", target.addr);

		connected.add(new RapidoidChannel(socketChannel, true, target.protocol, target.holder,
			target.reconnecting, target.state));

	} catch (ConnectException e) {
		retryConnecting(target);
	}
}
 

public static boolean checkIfRemoteEndpointAccessible(String host, int port) {
  try {
    Socket discover = new Socket();
    discover.setSoTimeout(1000);
    discover.connect(new InetSocketAddress(host, port), 1000);
    discover.close();
    return true;
  } catch (ConnectException cne) {
    // end point is not accessible
    if (LOG.isDebugEnabled()) {
      LOG.debug("Remote endpoint '" + host + ":" + port + "' is not accessible " +
          "(might be initializing): " + cne.getMessage());
    }
    return false;
  } catch (IOException ioe) {
    // end point is not accessible
    if (LOG.isDebugEnabled()) {
      LOG.debug("Remote endpoint '" + host + ":" + port + "' is not accessible " +
          "(might be initializing): " + ioe.getMessage());
    }
    return false;
  }
}
 

private static void connect(SelectionKey k) throws IOException {
    NioClientSocketChannel ch = (NioClientSocketChannel) k.attachment();
    try {
        if (ch.channel.finishConnect()) {
            k.cancel();
            if (ch.timoutTimer != null) {
                ch.timoutTimer.cancel();
            }
            ch.worker.register(ch, ch.connectFuture);
        }
    } catch (ConnectException e) {
        ConnectException newE = new ConnectException(e.getMessage() + ": " + ch.requestedRemoteAddress);
        newE.setStackTrace(e.getStackTrace());
        throw newE;
    }
}
 

/**
 * Take an IOException and a URI, wrap it where possible with
 * something that includes the URI
 *
 * @param dest target URI
 * @param operation operation
 * @param exception the caught exception.
 * @return an exception to throw
 */
public static IOException wrapException(final String dest,
                                        final String operation,
                                        final IOException exception) {
  String action = operation + " " + dest;
  String xref = null;

  if (exception instanceof ConnectException) {
    xref = "ConnectionRefused";
  } else if (exception instanceof UnknownHostException) {
    xref = "UnknownHost";
  } else if (exception instanceof SocketTimeoutException) {
    xref = "SocketTimeout";
  } else if (exception instanceof NoRouteToHostException) {
    xref = "NoRouteToHost";
  }
  String msg = action
               + " failed on exception: "
               + exception;
  if (xref != null) {
     msg = msg + ";" + see(xref);
  }
  return wrapWithMessage(exception, msg);
}
 

/**
 * This method handles a connection error that was caused while connecting the gateway to the backend.
 *
 * @param error the connection error to be handled
 * @return a new ConnectorException
 */
public static ConnectorException handleConnectionError(Throwable error) {
    ConnectorException ce = null;
    if (error instanceof UnknownHostException || error instanceof ConnectException || error instanceof NoRouteToHostException) {
        ce = new ConnectorException("Unable to connect to backend", error); //$NON-NLS-1$
        ce.setStatusCode(502); // BAD GATEWAY
    } else if (error instanceof InterruptedIOException || error instanceof java.util.concurrent.TimeoutException) {
        ce = new ConnectorException("Connection to backend terminated" + error.getMessage(), error); //$NON-NLS-1$
        ce.setStatusCode(504); // GATEWAY TIMEOUT

    }
    if (ce != null) {
        return ce;
    } else {
        return new ConnectorException(error.getMessage(), error);
    }
}
 

public Profile createProfile(String name) {
	CreateProfile request = new CreateProfile();
	CreateProfileResponse response = new CreateProfileResponse();

	request.setName(name);

	try {
		response = (CreateProfileResponse) soap.createSOAPMediaRequest(request, response, true);
	}
	catch (SOAPException | ConnectException e) {
		e.printStackTrace();
		return null;
	}

	if (response == null) {
		return null;
	}

	return response.getProfile();
}
 

@Override
public boolean reconcileConfiguration(final OvsdbConnectionManager connectionManagerOfDevice) {
    boolean result = false;
    connectionAttempt.incrementAndGet();
    InstanceIdentifier<Node> ndIid = (InstanceIdentifier<Node>) nodeIid;
    OvsdbNodeAugmentation ovsdbNode = (OvsdbNodeAugmentation)configData;

    LOG.info("Retry({}) connection to Ovsdb Node {} ", connectionAttempt.get(), ovsdbNode.getConnectionInfo());
    OvsdbClient client = null;
    try {
        client = connectionManagerOfDevice.connect(ndIid, ovsdbNode);
        if (client != null) {
            LOG.info("Successfully connected to Ovsdb Node {} ", ovsdbNode.getConnectionInfo());
            result = true;
        } else {
            LOG.warn("Connection retry({}) failed for {}.",
                    connectionAttempt.get(), ovsdbNode.getConnectionInfo());
        }
    } catch (UnknownHostException | ConnectException e) {
        LOG.warn("Connection retry({}) failed with exception. ",connectionAttempt.get(), e);
    }
    return result;
}
 

private String causeToString(Throwable t) {
    final Context c = getApplication();
    if (t instanceof InvalidSessionResourceException) {
        return c.getString(R.string.invalid_session_resource);
    }
    if (t instanceof EndpointNotFoundException) {
        return c.getString(R.string.endpoint_not_found);
    }
    if (t instanceof ConnectException) {
        return c.getString(R.string.unable_to_connect);
    }
    if (t instanceof SocketTimeoutException) {
        return c.getString(R.string.timeout_reached);
    }
    if (t instanceof SSLHandshakeException) {
        return c.getString(R.string.unable_to_establish_secure_connection);
    }
    if (t instanceof SSLPeerUnverifiedException) {
        return c.getString(R.string.unable_to_verify_service_identity);
    }
    throw new IllegalArgumentException();
}
 

@Override
public void onError(Throwable throwable) {
    if (throwable instanceof StatusRuntimeException) {
        final StatusRuntimeException sre = (StatusRuntimeException) throwable;
        if (sre.getStatus().getCause() instanceof ConnectException) {
            log.warn("{} is unreachable ({})",
                     deviceId, sre.getCause().getMessage());
        } else {
            log.warn("Error on StreamChannel RPC for {}: {}",
                     deviceId, throwable.getMessage());
        }
        log.debug("", throwable);
    } else {
        log.error(format("Exception on StreamChannel RPC for %s",
                         deviceId), throwable);
    }
    streamChannelManager.teardown();
}
 

/**
 * Write to channel.
 *
 * @param channel the channel
 * @param magicBytes the magic bytes
 * @param pathOrCommand the path or command
 * @param attachment the attachment
 */
private void writeToChannel(Channel channel, String[] magicBytes, Object pathOrCommand, Object attachment) throws ConnectException {
	long firstAttempt = System.currentTimeMillis();
	long timeOut = RemotingConstants.TEN * RemotingConstants.THOUSAND;
	while (!channel.isOpen() || !channel.isActive()) {
		if (System.currentTimeMillis() - firstAttempt >= timeOut) {
			try {
				throw new TimeoutException();
			} catch (TimeoutException e) {
				logger.error("Waited for 10 sec for connection reattempt to JumbuneAgent, but failed to connect", e);
			}
			break;
		}
	}
	if (!channel.isActive()) {
		logger.warn("channel #" + channel.hashCode() + " still disconnected, about to write on disconnected Channel");
	}
	if (attachment != null && attachment instanceof CyclicBarrier) {
		channel.attr(RemotingConstants.barrierKey).set((CyclicBarrier)attachment);
	}else if (attachment != null) {
		channel.attr(RemotingConstants.handlerKey).set((ChannelInboundHandler)attachment);
	}
	channel.write(Unpooled.wrappedBuffer(magicBytes[0].getBytes(), magicBytes[1].getBytes(), magicBytes[2].getBytes()));
	channel.write(pathOrCommand);
	channel.flush();
}
 

@Override
public Observable<?> call(Observable<? extends Throwable> observable) {
    return observable
            .zipWith(Observable.range(1, count + 1), new Func2<Throwable, Integer, Wrapper>() {
                @Override
                public Wrapper call(Throwable throwable, Integer integer) {
                    //压缩规则 合并后的结果是一个Observable<Wrapper>
                    return new Wrapper(throwable, integer);
                }
            }).flatMap(new Func1<Wrapper, Observable<?>>() {
                @Override
                public Observable<?> call(Wrapper wrapper) {
                    //转换规则
                    if ((wrapper.throwable instanceof ConnectException
                            || wrapper.throwable instanceof SocketTimeoutException
                            || wrapper.throwable instanceof TimeoutException)
                            && wrapper.index < count + 1) { //如果超出重试次数也抛出错误，否则默认是会进入onCompleted
                        return Observable.timer(delay + (wrapper.index - 1) * increaseDelay, TimeUnit.MILLISECONDS);

                    }
                    return Observable.error(wrapper.throwable);
                }
            });
}
 

@Test
public void testNewClientChannel_UnableToConnect_Blocking() throws Exception {

	manager = new ChannelManagerImpl(2, 0);
	manager.init();

	CountDownLatch trigger = new CountDownLatch(1);
	clientHandler.onChannelClosed(trigger);

	try {
		clientChannel = manager.newClientChannel("localhost", 19876, clientHandler);
		fail("expected exception");
	} catch (MqttInvocationException e) {
	}

	assertTrue(trigger.await(1000, TimeUnit.SECONDS));

	clientHandler.assertLastChannelClosedCause(ConnectException.class);
}
 

public boolean setImagingSettings(String videoSourceToken, ImagingSettings20 imagingSettings) {
	if (videoSourceToken == null) {
		return false;
	}

	SetImagingSettings request = new SetImagingSettings();
	SetImagingSettingsResponse response = new SetImagingSettingsResponse();

	request.setVideoSourceToken(videoSourceToken);
	request.setImagingSettings(imagingSettings);

	try {
		response = (SetImagingSettingsResponse) soap.createSOAPImagingRequest(request, response, true);
	}
	catch (SOAPException | ConnectException e) {
		e.printStackTrace();
		return false;
	}

	if (response == null) {
		return false;
	}

	return true;
}
 

public OvsdbClient connect(final InstanceIdentifier<Node> iid,
        final OvsdbNodeAugmentation ovsdbNode) throws UnknownHostException, ConnectException {
    LOG.info("Connecting to {}", SouthboundUtil.connectionInfoToString(ovsdbNode.getConnectionInfo()));

    // TODO handle case where we already have a connection
    // TODO use transaction chains to handle ordering issues between disconnected
    // TODO and connected when writing to the operational store
    InetAddress ip = SouthboundMapper.createInetAddress(ovsdbNode.getConnectionInfo().getRemoteIp());
    OvsdbClient client = ovsdbConnection.connect(ip,
            ovsdbNode.getConnectionInfo().getRemotePort().getValue().toJava());
    // For connections from the controller to the ovs instance, the library doesn't call
    // this method for us
    if (client != null) {
        putInstanceIdentifier(ovsdbNode.getConnectionInfo(), iid.firstIdentifierOf(Node.class));
        OvsdbConnectionInstance ovsdbConnectionInstance = connectedButCallBacksNotRegistered(client);
        ovsdbConnectionInstance.setOvsdbNodeAugmentation(ovsdbNode);

        // Register Cluster Ownership for ConnectionInfo
        registerEntityForOwnership(ovsdbConnectionInstance);
    } else {
        LOG.warn("Failed to connect to OVSDB Node {}", ovsdbNode.getConnectionInfo());
    }
    return client;
}
 

public String getSnapshotUri(String profileToken) throws SOAPException, ConnectException {
	GetSnapshotUri request = new GetSnapshotUri();
	GetSnapshotUriResponse response = new GetSnapshotUriResponse();

	request.setProfileToken(profileToken);

	try {
		response = (GetSnapshotUriResponse) soap.createSOAPMediaRequest(request, response, true);
	}
	catch (SOAPException | ConnectException e) {
		throw e;
	}
	
	if (response == null || response.getMediaUri() == null) {
		return null;
	}
	
	return onvifDevice.replaceLocalIpWithProxyIp(response.getMediaUri().getUri());
}
 

/** Does all the work necessary to build a full HTTP or HTTPS connection on a raw socket. */
private void connectSocket(int connectTimeout, int readTimeout) throws IOException {
  Proxy proxy = route.proxy();
  Address address = route.address();

  rawSocket = proxy.type() == Proxy.Type.DIRECT || proxy.type() == Proxy.Type.HTTP
      ? address.socketFactory().createSocket()
      : new Socket(proxy);

  rawSocket.setSoTimeout(readTimeout);
  try {
    Platform.get().connectSocket(rawSocket, route.socketAddress(), connectTimeout);
  } catch (ConnectException e) {
    ConnectException ce = new ConnectException("Failed to connect to " + route.socketAddress());
    ce.initCause(e);
    throw ce;
  }
  source = Okio.buffer(Okio.source(rawSocket));
  sink = Okio.buffer(Okio.sink(rawSocket));
}
 

public String explanationText(Mail mail, Exception ex) {
    StringWriter sout = new StringWriter();
    PrintWriter out = new PrintWriter(sout, true);
    out.println("Hi. This is the James mail server at " + resolveMachineName() + ".");
    out.println("I'm afraid I wasn't able to deliver your message to the following addresses.");
    out.println("This is a permanent error; I've given up. Sorry it didn't work out. Below");
    out.println("I include the list of recipients and the reason why I was unable to deliver");
    out.println("your message.");
    out.println();
    for (MailAddress mailAddress : mail.getRecipients()) {
        out.println(mailAddress);
    }
    if (ex instanceof MessagingException) {
        if (((MessagingException) ex).getNextException() == null) {
            out.println(sanitizeExceptionMessage(ex));
        } else {
            Exception ex1 = ((MessagingException) ex).getNextException();
            if (ex1 instanceof SendFailedException) {
                out.println("Remote mail server told me: " + sanitizeExceptionMessage(ex1));
            } else if (ex1 instanceof UnknownHostException) {
                out.println("Unknown host: " + sanitizeExceptionMessage(ex1));
                out.println("This could be a DNS server error, a typo, or a problem with the recipient's mail server.");
            } else if (ex1 instanceof ConnectException) {
                // Already formatted as "Connection timed out: connect"
                out.println(sanitizeExceptionMessage(ex1));
            } else if (ex1 instanceof SocketException) {
                out.println("Socket exception: " + sanitizeExceptionMessage(ex1));
            } else {
                out.println(sanitizeExceptionMessage(ex1));
            }
        }
    }
    out.println();
    return sout.toString();
}
 

public void test_newSocket_connection_refused() throws Exception {
    try {
        new Socket("localhost", 80);
        fail("connection should have been refused");
    } catch (ConnectException expected) {
    }
}
 

private void handleException(MessagingException e){
	if (e instanceof AuthenticationFailedException) {
		lastError = ErrorTyp.AUTHENTICATION;
	} else if (e.getNextException() instanceof UnknownHostException
		|| e.getNextException() instanceof ConnectException) {
		lastError = ErrorTyp.CONNECTION;
	} else if (e instanceof AddressException) {
		lastError = ErrorTyp.ADDRESS;
	}
}
 

private CompletableFuture<LocalRaftServerProtocol> getServer(MemberId memberId) {
  LocalRaftServerProtocol server = server(memberId);
  if (server != null) {
    return Futures.completedFuture(server);
  } else {
    return Futures.exceptionalFuture(new ConnectException());
  }
}
 

private static boolean isSecurePort(String hostname, int port, int depth) 
        throws IOException, ConnectException, SocketTimeoutException {
    // Open the socket with a short timeout for connects and reads.
    Socket socket = new Socket();
    try {
        Logger.getLogger("glassfish-socket-connect-diagnostic").log(Level.FINE, "Using socket.connect", new Exception());
        socket.connect(new InetSocketAddress(hostname, port), PORT_CHECK_TIMEOUT);
        socket.setSoTimeout(PORT_CHECK_TIMEOUT);
    } catch(SocketException ex) { // this could be bug 70020 due to SOCKs proxy not having localhost
        String socksNonProxyHosts = System.getProperty("socksNonProxyHosts");
        if(socksNonProxyHosts != null && socksNonProxyHosts.indexOf("localhost") < 0) {
            String localhost = socksNonProxyHosts.length() > 0 ? "|localhost" : "localhost";
            System.setProperty("socksNonProxyHosts",  socksNonProxyHosts + localhost);
            ConnectException ce = new ConnectException();
            ce.initCause(ex);
            throw ce; //status unknow at this point
            //next call, we'll be ok and it will really detect if we are secure or not
        }
    }
    //This is the test query used to ping the server in an attempt to
    //determine if it is secure or not.
    InputStream is = socket.getInputStream();        
    String testQuery = "GET / HTTP/1.0";
    PrintWriter pw = new PrintWriter(socket.getOutputStream());
    pw.println(testQuery);
    pw.println();
    pw.flush();
    byte[] respArr = new byte[1024];
    boolean isSecure = true;
    while (is.read(respArr) != -1) {
        String resp = new String(respArr);
        if (checkHelper(resp) == false) {
            isSecure = false;
            break;
        }
    }
    // Close the socket
    socket.close();
    return isSecure;
}
 

@Test
public void testNoConnectionToDatabase() throws Exception {
    NotifyBuilder notify = new NotifyBuilder(context).whenDone(1).create();

    RouteBuilder rb = new RouteBuilder() {
        @Override
        public void configure() throws Exception {
            interceptSendToEndpoint("sql:*")
                .skipSendToOriginalEndpoint()
                .throwException(new ConnectException("Cannot connect to the database"));
        }
    };

    // adviseWith enhances our route by adding the interceptor from the route builder
    // this allows us here directly in the unit test to add interceptors so we can simulate the connection failure
    context.getRouteDefinition("partnerToDB").adviceWith(context, rb);

    // start Camel after advice
    context.start();

    // there should be 0 row in the database when we start
    int rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(0, rows);

    String xml = "<?xml version=\"1.0\"?><partner id=\"123\"><date>201702250815</date><code>200</code><time>4387</time></partner>";
    template.sendBody("activemq:queue:partners", xml);

    // wait for the route to complete one message
    assertTrue(notify.matches(10, TimeUnit.SECONDS));

    // data not inserted so there should be 0 rows
    rows = jdbc.queryForObject("select count(*) from partner_metric", Integer.class);
    assertEquals(0, rows);
}
 

public static PipelineStateJson getEdgePipelineState(
    PipelineConfiguration pipelineConfiguration
) throws PipelineException {
  String pipelineId = pipelineConfiguration.getPipelineId();
  PipelineConfigBean pipelineConfigBean =  PipelineBeanCreator.get()
      .create(pipelineConfiguration, new ArrayList<>(), null);
  if (!pipelineConfigBean.executionMode.equals(ExecutionMode.EDGE)) {
    throw new PipelineException(ContainerError.CONTAINER_01600, pipelineConfigBean.executionMode);
  }

  Response response = null;
  try {
    response = ClientBuilder.newClient()
        .target(pipelineConfigBean.edgeHttpUrl + "/rest/v1/pipeline/" + pipelineId + "/status")
        .request()
        .get();
    if (response.getStatus() == Response.Status.OK.getStatusCode()) {
      return response.readEntity(PipelineStateJson.class);
    } else {
      return null;
    }
  } catch (ProcessingException ex) {
    if (ex.getCause() instanceof ConnectException) {
      throw new PipelineException(ContainerError.CONTAINER_01602, pipelineConfigBean.edgeHttpUrl, ex);
    }
    throw ex;
  }
  finally {
    if (response != null) {
      response.close();
    }
  }
}