@Test public void ensureDefaultRequestTimeoutCorrect() { whenDefaultStringConfig(); when(configContext.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT)).thenReturn(1000); when(configContext.getInteger(eq("connectTimeout"), Mockito.anyInt())).thenReturn(1000); new HttpSink().configure(configContext); verify(configContext).getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); }
@Override public final void configure(final Context context) { String configuredEndpoint = context.getString("endpoint", ""); LOG.info("Read endpoint URL from configuration : " + configuredEndpoint); try { endpointUrl = new URL(configuredEndpoint); } catch (MalformedURLException e) { throw new IllegalArgumentException("Endpoint URL invalid", e); } connectTimeout = context.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); if (connectTimeout <= 0) { throw new IllegalArgumentException( "Connect timeout must be a non-zero and positive"); } LOG.info("Using connect timeout : " + connectTimeout); requestTimeout = context.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); if (requestTimeout <= 0) { throw new IllegalArgumentException( "Request timeout must be a non-zero and positive"); } LOG.info("Using request timeout : " + requestTimeout); acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER); LOG.info("Using Accept header value : " + acceptHeader); contentTypeHeader = context.getString("contentTypeHeader", DEFAULT_CONTENT_TYPE); LOG.info("Using Content-Type header value : " + contentTypeHeader); defaultBackoff = context.getBoolean("defaultBackoff", true); LOG.info("Channel backoff by default is " + defaultBackoff); defaultRollback = context.getBoolean("defaultRollback", true); LOG.info("Transaction rollback by default is " + defaultRollback); defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics", false); LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics); parseConfigOverrides("backoff", context, backoffOverrides); parseConfigOverrides("rollback", context, rollbackOverrides); parseConfigOverrides("incrementMetrics", context, incrementMetricsOverrides); if (this.sinkCounter == null) { this.sinkCounter = new SinkCounter(this.getName()); } connectionBuilder = new ConnectionBuilder(); }
HttpSink extends AbstractSink implements Configurable { @Override public final void configure(final Context context) { String configuredEndpoint = context.getString("endpoint", ""); LOG.info("Read endpoint URL from configuration : " + configuredEndpoint); try { endpointUrl = new URL(configuredEndpoint); } catch (MalformedURLException e) { throw new IllegalArgumentException("Endpoint URL invalid", e); } connectTimeout = context.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); if (connectTimeout <= 0) { throw new IllegalArgumentException( "Connect timeout must be a non-zero and positive"); } LOG.info("Using connect timeout : " + connectTimeout); requestTimeout = context.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); if (requestTimeout <= 0) { throw new IllegalArgumentException( "Request timeout must be a non-zero and positive"); } LOG.info("Using request timeout : " + requestTimeout); acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER); LOG.info("Using Accept header value : " + acceptHeader); contentTypeHeader = context.getString("contentTypeHeader", DEFAULT_CONTENT_TYPE); LOG.info("Using Content-Type header value : " + contentTypeHeader); defaultBackoff = context.getBoolean("defaultBackoff", true); LOG.info("Channel backoff by default is " + defaultBackoff); defaultRollback = context.getBoolean("defaultRollback", true); LOG.info("Transaction rollback by default is " + defaultRollback); defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics", false); LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics); parseConfigOverrides("backoff", context, backoffOverrides); parseConfigOverrides("rollback", context, rollbackOverrides); parseConfigOverrides("incrementMetrics", context, incrementMetricsOverrides); if (this.sinkCounter == null) { this.sinkCounter = new SinkCounter(this.getName()); } connectionBuilder = new ConnectionBuilder(); } }
HttpSink extends AbstractSink implements Configurable { @Override public final void configure(final Context context) { String configuredEndpoint = context.getString("endpoint", ""); LOG.info("Read endpoint URL from configuration : " + configuredEndpoint); try { endpointUrl = new URL(configuredEndpoint); } catch (MalformedURLException e) { throw new IllegalArgumentException("Endpoint URL invalid", e); } connectTimeout = context.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); if (connectTimeout <= 0) { throw new IllegalArgumentException( "Connect timeout must be a non-zero and positive"); } LOG.info("Using connect timeout : " + connectTimeout); requestTimeout = context.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); if (requestTimeout <= 0) { throw new IllegalArgumentException( "Request timeout must be a non-zero and positive"); } LOG.info("Using request timeout : " + requestTimeout); acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER); LOG.info("Using Accept header value : " + acceptHeader); contentTypeHeader = context.getString("contentTypeHeader", DEFAULT_CONTENT_TYPE); LOG.info("Using Content-Type header value : " + contentTypeHeader); defaultBackoff = context.getBoolean("defaultBackoff", true); LOG.info("Channel backoff by default is " + defaultBackoff); defaultRollback = context.getBoolean("defaultRollback", true); LOG.info("Transaction rollback by default is " + defaultRollback); defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics", false); LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics); parseConfigOverrides("backoff", context, backoffOverrides); parseConfigOverrides("rollback", context, rollbackOverrides); parseConfigOverrides("incrementMetrics", context, incrementMetricsOverrides); if (this.sinkCounter == null) { this.sinkCounter = new SinkCounter(this.getName()); } connectionBuilder = new ConnectionBuilder(); }  }
HttpSink extends AbstractSink implements Configurable { @Override public final void configure(final Context context) { String configuredEndpoint = context.getString("endpoint", ""); LOG.info("Read endpoint URL from configuration : " + configuredEndpoint); try { endpointUrl = new URL(configuredEndpoint); } catch (MalformedURLException e) { throw new IllegalArgumentException("Endpoint URL invalid", e); } connectTimeout = context.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); if (connectTimeout <= 0) { throw new IllegalArgumentException( "Connect timeout must be a non-zero and positive"); } LOG.info("Using connect timeout : " + connectTimeout); requestTimeout = context.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); if (requestTimeout <= 0) { throw new IllegalArgumentException( "Request timeout must be a non-zero and positive"); } LOG.info("Using request timeout : " + requestTimeout); acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER); LOG.info("Using Accept header value : " + acceptHeader); contentTypeHeader = context.getString("contentTypeHeader", DEFAULT_CONTENT_TYPE); LOG.info("Using Content-Type header value : " + contentTypeHeader); defaultBackoff = context.getBoolean("defaultBackoff", true); LOG.info("Channel backoff by default is " + defaultBackoff); defaultRollback = context.getBoolean("defaultRollback", true); LOG.info("Transaction rollback by default is " + defaultRollback); defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics", false); LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics); parseConfigOverrides("backoff", context, backoffOverrides); parseConfigOverrides("rollback", context, rollbackOverrides); parseConfigOverrides("incrementMetrics", context, incrementMetricsOverrides); if (this.sinkCounter == null) { this.sinkCounter = new SinkCounter(this.getName()); } connectionBuilder = new ConnectionBuilder(); }  @Override final void configure(final Context context); @Override final void start(); @Override final void stop(); @Override final Status process(); }
HttpSink extends AbstractSink implements Configurable { @Override public final void configure(final Context context) { String configuredEndpoint = context.getString("endpoint", ""); LOG.info("Read endpoint URL from configuration : " + configuredEndpoint); try { endpointUrl = new URL(configuredEndpoint); } catch (MalformedURLException e) { throw new IllegalArgumentException("Endpoint URL invalid", e); } connectTimeout = context.getInteger("connectTimeout", DEFAULT_CONNECT_TIMEOUT); if (connectTimeout <= 0) { throw new IllegalArgumentException( "Connect timeout must be a non-zero and positive"); } LOG.info("Using connect timeout : " + connectTimeout); requestTimeout = context.getInteger("requestTimeout", DEFAULT_REQUEST_TIMEOUT); if (requestTimeout <= 0) { throw new IllegalArgumentException( "Request timeout must be a non-zero and positive"); } LOG.info("Using request timeout : " + requestTimeout); acceptHeader = context.getString("acceptHeader", DEFAULT_ACCEPT_HEADER); LOG.info("Using Accept header value : " + acceptHeader); contentTypeHeader = context.getString("contentTypeHeader", DEFAULT_CONTENT_TYPE); LOG.info("Using Content-Type header value : " + contentTypeHeader); defaultBackoff = context.getBoolean("defaultBackoff", true); LOG.info("Channel backoff by default is " + defaultBackoff); defaultRollback = context.getBoolean("defaultRollback", true); LOG.info("Transaction rollback by default is " + defaultRollback); defaultIncrementMetrics = context.getBoolean("defaultIncrementMetrics", false); LOG.info("Incrementing metrics by default is " + defaultIncrementMetrics); parseConfigOverrides("backoff", context, backoffOverrides); parseConfigOverrides("rollback", context, rollbackOverrides); parseConfigOverrides("incrementMetrics", context, incrementMetricsOverrides); if (this.sinkCounter == null) { this.sinkCounter = new SinkCounter(this.getName()); } connectionBuilder = new ConnectionBuilder(); }  @Override final void configure(final Context context); @Override final void start(); @Override final void stop(); @Override final Status process();  }