### Question:
ConsumeKafka extends AbstractProcessor { @OnStopped public void close() { final ConsumerPool pool = consumerPool; consumerPool = null; if (pool != null) { pool.close(); } }  @Override Set<Relationship> getRelationships(); @OnStopped void close(); @OnUnscheduled void interruptActiveThreads(); @Override void onTrigger(ProcessContext context, ProcessSession session);  }### Answer:
@Test public void validateGetAllMessages() throws Exception { String groupName = "validateGetAllMessages"; when(mockConsumerPool.obtainConsumer(anyObject())).thenReturn(mockLease); when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE); when(mockLease.commit()).thenReturn(Boolean.TRUE); ConsumeKafka proc = new ConsumeKafka() { @Override protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) { return mockConsumerPool; } }; final TestRunner runner = TestRunners.newTestRunner(proc); runner.setValidateExpressionUsage(false); runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234"); runner.setProperty(ConsumeKafka.TOPICS, "foo,bar"); runner.setProperty(ConsumeKafka.GROUP_ID, groupName); runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET, ConsumeKafka.OFFSET_EARLIEST); runner.run(1, false); verify(mockConsumerPool, times(1)).obtainConsumer(anyObject()); verify(mockLease, times(3)).continuePolling(); verify(mockLease, times(2)).poll(); verify(mockLease, times(1)).commit(); verify(mockLease, times(1)).close(); verifyNoMoreInteractions(mockConsumerPool); verifyNoMoreInteractions(mockLease); }
@Test public void validateGetErrorMessages() throws Exception { String groupName = "validateGetErrorMessages"; when(mockConsumerPool.obtainConsumer(anyObject())).thenReturn(mockLease); when(mockLease.continuePolling()).thenReturn(true, false); when(mockLease.commit()).thenReturn(Boolean.FALSE); ConsumeKafka proc = new ConsumeKafka() { @Override protected ConsumerPool createConsumerPool(final ProcessContext context, final ComponentLog log) { return mockConsumerPool; } }; final TestRunner runner = TestRunners.newTestRunner(proc); runner.setValidateExpressionUsage(false); runner.setProperty(KafkaProcessorUtils.BOOTSTRAP_SERVERS, "0.0.0.0:1234"); runner.setProperty(ConsumeKafka.TOPICS, "foo,bar"); runner.setProperty(ConsumeKafka.GROUP_ID, groupName); runner.setProperty(ConsumeKafka.AUTO_OFFSET_RESET, ConsumeKafka.OFFSET_EARLIEST); runner.run(1, false); verify(mockConsumerPool, times(1)).obtainConsumer(anyObject()); verify(mockLease, times(2)).continuePolling(); verify(mockLease, times(1)).poll(); verify(mockLease, times(1)).commit(); verify(mockLease, times(1)).close(); verifyNoMoreInteractions(mockConsumerPool); verifyNoMoreInteractions(mockLease); }