BulkManager implements Runnable { @VisibleForTesting @SuppressWarnings({"checkstyle:designforextension"}) protected boolean getDataFromKafka(final StringBuilder buffer) { ConsumerData cd = simpleConsumerManager.readItem(); if (cd == null) { return false; } String message = cd.getMessage(); if (!isValidJsonObject(message)) { return false; } if (buffer.length() > 0) { buffer.append("\r\n"); } metrics.readKafkaItemFromConsumer.mark(); buffer.append(message); return true; } 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);  }
@Test public void getDataFromKafka_should_return_just_item_if_buffer_empty() { SimpleConsumerManager scm = mock(SimpleConsumerManager.class); ConsumerData cd = mock(ConsumerData.class); when(cd.getMessage()).thenReturn("{\"DATA\":\"1\"}"); when(scm.readItem()).thenReturn(cd); Metrics metrics = new Metrics(); BulkManager bm = new BulkManager(null, scm, null, metrics); StringBuilder sb = new StringBuilder(); assertTrue(bm.getDataFromKafka(sb)); assertEquals("{\"DATA\":\"1\"}", sb.toString()); assertEquals(1, metrics.readKafkaItemFromConsumer.getCount()); }
@Test public void getDataFromKafka_should_concatenate_item_to_not_empty_stringbuilder() { SimpleConsumerManager scm = mock(SimpleConsumerManager.class); ConsumerData cd = mock(ConsumerData.class); when(cd.getMessage()).thenReturn("{\"DATA\":\"1\"}"); when(scm.readItem()).thenReturn(cd); Metrics metrics = new Metrics(); BulkManager bm = new BulkManager(null, scm, null, metrics); StringBuilder sb = new StringBuilder(); sb.append("EXISTING"); assertTrue(bm.getDataFromKafka(sb)); assertEquals("EXISTING\r\n{\"DATA\":\"1\"}", sb.toString()); }
@Test public void getDataFromKafka_should_drop_message_that_is_not_a_json_object() { SimpleConsumerManager scm = mock(SimpleConsumerManager.class); ConsumerData cd = mock(ConsumerData.class); when(cd.getMessage()).thenReturn("DATA"); when(scm.readItem()).thenReturn(cd); Metrics metrics = new Metrics(); BulkManager bm = new BulkManager(null, scm, null, metrics); StringBuilder sb = new StringBuilder(); sb.append("EXISTING"); assertFalse(bm.getDataFromKafka(sb)); assertEquals("EXISTING", sb.toString()); }
@Test public void getDataFromKafka_should_not_concatenate_if_no_item_returned() { SimpleConsumerManager scm = mock(SimpleConsumerManager.class); Metrics metrics = new Metrics(); BulkManager bm = new BulkManager(null, scm, null, metrics); StringBuilder sb = new StringBuilder(); sb.append("EXISTING"); assertFalse(bm.getDataFromKafka(sb)); assertEquals("EXISTING", sb.toString()); assertEquals(0, metrics.readKafkaItemFromConsumer.getCount()); }