@Test
public void testErrorCounters() throws InterruptedException, EventDeliveryException {

  context.put(TOPICS, topic0);
  context.put(BATCH_SIZE, "1");
  kafkaSource.configure(context);
  ChannelProcessor cp = Mockito.mock(ChannelProcessor.class);
  doThrow(new ChannelException("dummy")).doThrow(new RuntimeException("dummy"))
      .when(cp).processEventBatch(any(List.class));
  kafkaSource.setChannelProcessor(cp);
  startKafkaSource();
  Thread.sleep(500L);
  kafkaServer.produce(topic0, "", "hello, world");
  Thread.sleep(500L);
  kafkaSource.doProcess();
  kafkaSource.doProcess();
  SourceCounter sc = (SourceCounter) Whitebox.getInternalState(kafkaSource, "counter");
  Assert.assertEquals(1, sc.getChannelWriteFail());
  Assert.assertEquals(1, sc.getEventReadFail());
  kafkaSource.stop();
}