class 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();
  }

}