@Test public void should_not_send_if_buffer_empty() throws Exception { BulkManager bm = new BulkManager(null, null, null, null); bm.send(new BulkReadValues(0, new StringBuilder().toString())); }
@SuppressWarnings("checkstyle:designforextension") protected void send(final BulkReadValues readValues) throws InterruptedException { log.debug("send()"); HttpResponse response = null; try { if (readValues.getData().equals("")) { return; } final Timer.Context context = metrics.bulkPostTime.time(); response = post(readValues.getData()); context.stop(); int statusCode = response.getStatusLine().getStatusCode(); if (statusCode == HttpStatus.SC_OK) { simpleConsumerManager.commit(); String body = EntityUtils.toString(response.getEntity()); metrics.sendSuccess.mark(); metrics.sentItems.mark(readValues.getItemsRead()); backoff.reset(); log.trace("Data successfully sent to ingestion endpoint: {}", body); log.debug("Data successfully sent to ingestion endpoint: hash {}", body.hashCode()); } else if (statusCode == HttpStatus.SC_REQUEST_TOO_LONG || statusCode == SC_TOO_MANY_REQUESTS) { simpleConsumerManager.reset(); long ttl = Long.parseLong( response.getFirstHeader("X-Ingestion-Data-RateLimit-Reset-Ttl").getValue()); long ttlms = TimeUnit.SECONDS.toMillis(ttl); EntityUtils.consume(response.getEntity()); log.info("Rate limited, waiting until limits reset at {}", new Date(ttlms)); metrics.sendRateLimit.mark(); backoff.waitUntil(ttl); } else if (statusCode >= HttpStatus.SC_BAD_REQUEST) { simpleConsumerManager.reset(); EntityUtils.consume(response.getEntity()); log.error("Error code returned from ingestion endpoint, status = {}", statusCode); metrics.sendError.mark(); backoff.exponentialBackoff(); } } catch (URISyntaxException | IOException | RuntimeException e) { simpleConsumerManager.reset(); if (response != null) { try { EntityUtils.consume(response.getEntity()); } catch (IOException e1) { log.error("Couldn't consume response to close it", e); } } log.error("Could not connect to ingestion endpoint", e); metrics.sendException.mark(); backoff.linearBackoff(); } }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") protected void send(final BulkReadValues readValues) throws InterruptedException { log.debug("send()"); HttpResponse response = null; try { if (readValues.getData().equals("")) { return; } final Timer.Context context = metrics.bulkPostTime.time(); response = post(readValues.getData()); context.stop(); int statusCode = response.getStatusLine().getStatusCode(); if (statusCode == HttpStatus.SC_OK) { simpleConsumerManager.commit(); String body = EntityUtils.toString(response.getEntity()); metrics.sendSuccess.mark(); metrics.sentItems.mark(readValues.getItemsRead()); backoff.reset(); log.trace("Data successfully sent to ingestion endpoint: {}", body); log.debug("Data successfully sent to ingestion endpoint: hash {}", body.hashCode()); } else if (statusCode == HttpStatus.SC_REQUEST_TOO_LONG || statusCode == SC_TOO_MANY_REQUESTS) { simpleConsumerManager.reset(); long ttl = Long.parseLong( response.getFirstHeader("X-Ingestion-Data-RateLimit-Reset-Ttl").getValue()); long ttlms = TimeUnit.SECONDS.toMillis(ttl); EntityUtils.consume(response.getEntity()); log.info("Rate limited, waiting until limits reset at {}", new Date(ttlms)); metrics.sendRateLimit.mark(); backoff.waitUntil(ttl); } else if (statusCode >= HttpStatus.SC_BAD_REQUEST) { simpleConsumerManager.reset(); EntityUtils.consume(response.getEntity()); log.error("Error code returned from ingestion endpoint, status = {}", statusCode); metrics.sendError.mark(); backoff.exponentialBackoff(); } } catch (URISyntaxException | IOException | RuntimeException e) { simpleConsumerManager.reset(); if (response != null) { try { EntityUtils.consume(response.getEntity()); } catch (IOException e1) { log.error("Couldn't consume response to close it", e); } } log.error("Could not connect to ingestion endpoint", e); metrics.sendException.mark(); backoff.linearBackoff(); } } }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") protected void send(final BulkReadValues readValues) throws InterruptedException { log.debug("send()"); HttpResponse response = null; try { if (readValues.getData().equals("")) { return; } final Timer.Context context = metrics.bulkPostTime.time(); response = post(readValues.getData()); context.stop(); int statusCode = response.getStatusLine().getStatusCode(); if (statusCode == HttpStatus.SC_OK) { simpleConsumerManager.commit(); String body = EntityUtils.toString(response.getEntity()); metrics.sendSuccess.mark(); metrics.sentItems.mark(readValues.getItemsRead()); backoff.reset(); log.trace("Data successfully sent to ingestion endpoint: {}", body); log.debug("Data successfully sent to ingestion endpoint: hash {}", body.hashCode()); } else if (statusCode == HttpStatus.SC_REQUEST_TOO_LONG || statusCode == SC_TOO_MANY_REQUESTS) { simpleConsumerManager.reset(); long ttl = Long.parseLong( response.getFirstHeader("X-Ingestion-Data-RateLimit-Reset-Ttl").getValue()); long ttlms = TimeUnit.SECONDS.toMillis(ttl); EntityUtils.consume(response.getEntity()); log.info("Rate limited, waiting until limits reset at {}", new Date(ttlms)); metrics.sendRateLimit.mark(); backoff.waitUntil(ttl); } else if (statusCode >= HttpStatus.SC_BAD_REQUEST) { simpleConsumerManager.reset(); EntityUtils.consume(response.getEntity()); log.error("Error code returned from ingestion endpoint, status = {}", statusCode); metrics.sendError.mark(); backoff.exponentialBackoff(); } } catch (URISyntaxException | IOException | RuntimeException e) { simpleConsumerManager.reset(); if (response != null) { try { EntityUtils.consume(response.getEntity()); } catch (IOException e1) { log.error("Couldn't consume response to close it", e); } } log.error("Could not connect to ingestion endpoint", e); metrics.sendException.mark(); backoff.linearBackoff(); } } BulkManager(final DataSiftConfig config,
                       final SimpleConsumerManager simpleConsumerManager,
                       final Backoff backoff,
                       final Metrics metrics); }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") protected void send(final BulkReadValues readValues) throws InterruptedException { log.debug("send()"); HttpResponse response = null; try { if (readValues.getData().equals("")) { return; } final Timer.Context context = metrics.bulkPostTime.time(); response = post(readValues.getData()); context.stop(); int statusCode = response.getStatusLine().getStatusCode(); if (statusCode == HttpStatus.SC_OK) { simpleConsumerManager.commit(); String body = EntityUtils.toString(response.getEntity()); metrics.sendSuccess.mark(); metrics.sentItems.mark(readValues.getItemsRead()); backoff.reset(); log.trace("Data successfully sent to ingestion endpoint: {}", body); log.debug("Data successfully sent to ingestion endpoint: hash {}", body.hashCode()); } else if (statusCode == HttpStatus.SC_REQUEST_TOO_LONG || statusCode == SC_TOO_MANY_REQUESTS) { simpleConsumerManager.reset(); long ttl = Long.parseLong( response.getFirstHeader("X-Ingestion-Data-RateLimit-Reset-Ttl").getValue()); long ttlms = TimeUnit.SECONDS.toMillis(ttl); EntityUtils.consume(response.getEntity()); log.info("Rate limited, waiting until limits reset at {}", new Date(ttlms)); metrics.sendRateLimit.mark(); backoff.waitUntil(ttl); } else if (statusCode >= HttpStatus.SC_BAD_REQUEST) { simpleConsumerManager.reset(); EntityUtils.consume(response.getEntity()); log.error("Error code returned from ingestion endpoint, status = {}", statusCode); metrics.sendError.mark(); backoff.exponentialBackoff(); } } catch (URISyntaxException | IOException | RuntimeException e) { simpleConsumerManager.reset(); if (response != null) { try { EntityUtils.consume(response.getEntity()); } catch (IOException e1) { log.error("Couldn't consume response to close it", e); } } log.error("Could not connect to ingestion endpoint", e); metrics.sendException.mark(); backoff.linearBackoff(); } } BulkManager(final DataSiftConfig config,
                       final SimpleConsumerManager simpleConsumerManager,
                       final Backoff backoff,
                       final Metrics metrics); final void setLogger(final Logger logger); final void shutdown(); final void run(); @SuppressWarnings("checkstyle:designforextension") HttpResponse post(final String data); }
BulkManager implements Runnable { @SuppressWarnings("checkstyle:designforextension") protected void send(final BulkReadValues readValues) throws InterruptedException { log.debug("send()"); HttpResponse response = null; try { if (readValues.getData().equals("")) { return; } final Timer.Context context = metrics.bulkPostTime.time(); response = post(readValues.getData()); context.stop(); int statusCode = response.getStatusLine().getStatusCode(); if (statusCode == HttpStatus.SC_OK) { simpleConsumerManager.commit(); String body = EntityUtils.toString(response.getEntity()); metrics.sendSuccess.mark(); metrics.sentItems.mark(readValues.getItemsRead()); backoff.reset(); log.trace("Data successfully sent to ingestion endpoint: {}", body); log.debug("Data successfully sent to ingestion endpoint: hash {}", body.hashCode()); } else if (statusCode == HttpStatus.SC_REQUEST_TOO_LONG || statusCode == SC_TOO_MANY_REQUESTS) { simpleConsumerManager.reset(); long ttl = Long.parseLong( response.getFirstHeader("X-Ingestion-Data-RateLimit-Reset-Ttl").getValue()); long ttlms = TimeUnit.SECONDS.toMillis(ttl); EntityUtils.consume(response.getEntity()); log.info("Rate limited, waiting until limits reset at {}", new Date(ttlms)); metrics.sendRateLimit.mark(); backoff.waitUntil(ttl); } else if (statusCode >= HttpStatus.SC_BAD_REQUEST) { simpleConsumerManager.reset(); EntityUtils.consume(response.getEntity()); log.error("Error code returned from ingestion endpoint, status = {}", statusCode); metrics.sendError.mark(); backoff.exponentialBackoff(); } } catch (URISyntaxException | IOException | RuntimeException e) { simpleConsumerManager.reset(); if (response != null) { try { EntityUtils.consume(response.getEntity()); } catch (IOException e1) { log.error("Couldn't consume response to close it", e); } } log.error("Could not connect to ingestion endpoint", e); metrics.sendException.mark(); backoff.linearBackoff(); } } BulkManager(final DataSiftConfig config,
                       final SimpleConsumerManager simpleConsumerManager,
                       final Backoff backoff,
                       final Metrics metrics); final void setLogger(final Logger logger); final void shutdown(); final void run(); @SuppressWarnings("checkstyle:designforextension") HttpResponse post(final String data);  }