KafkaWriter extends StreamingProcessor { @Override protected EventSet consume(ProcessingContext processingContext, EventSet eventSet) throws ProcessingException { final List<KeyedMessage<String, String>> messages = Lists.newArrayList(); try { eventSet.getEvents().forEach(event -> { KeyedMessage<String, String> convertedMessage = null; try { convertedMessage = convertEvent(event); } catch (ProcessingException e) { LOGGER.error("Error converting byte stream to event: ", e); throw new RuntimeException(e); } if (null != convertedMessage) { messages.add(convertedMessage); } }); } catch (final Exception e) { LOGGER.error("Error converting byte stream to event: ", e); throw new ProcessingException(e); } Lists.partition(messages, ingestionPoolSize).forEach(messageList -> getProducer().send(messageList)); return eventSet; }  @Override void initialize(String instanceId, Properties globalProperties, Properties properties,
            ComponentMetadata componentMetadata); @Override void destroy();  }
@Test public void testConsume() throws Exception { Random rnd = new Random(); ObjectMapper mapper = new ObjectMapper(); List<String> lst = Lists.newArrayList(); for (int i = 0; i < 5; i++) { lst.add("{\"firstName\":\"" + "xyz" + rnd.nextInt(10) + "\", \"lastName\":\"Doe" + rnd.nextInt(100) + "\"}"); } ProcessorTestBench processorTestBench = new ProcessorTestBench(false); List<EventSet> events = processorTestBench.runStreamingProcessor(processor, ImmutableList .of(EventSet.eventFromEventBuilder().events(ImmutableList .of(Event.builder().jsonNode(mapper.readTree(lst.get(0))).build(), Event.builder().jsonNode(mapper.readTree(lst.get(0))).build())).build(), EventSet.eventFromEventBuilder().events(ImmutableList .of(Event.builder().jsonNode(mapper.readTree(lst.get(0))).build(), Event.builder().jsonNode(mapper.readTree(lst.get(0))).build(), Event.builder().jsonNode(mapper.readTree(lst.get(0))).build())).build())); verify(producer, times(2)).send(anyList()); }