@Test public void validateGetAllMessagesPattern() throws Exception { String groupName = "validateGetAllMessagesPattern"; when(mockConsumerPool.obtainConsumer(anyObject(), anyObject())).thenReturn(mockLease); when(mockLease.continuePolling()).thenReturn(Boolean.TRUE, Boolean.TRUE, Boolean.FALSE); when(mockLease.commit()).thenReturn(Boolean.TRUE); runner.setProperty(ConsumeKafkaRecord_0_10.TOPICS, "(fo.*)|(ba)"); runner.setProperty(ConsumeKafkaRecord_0_10.TOPIC_TYPE, "pattern"); runner.setProperty(ConsumeKafkaRecord_0_10.GROUP_ID, groupName); runner.setProperty(ConsumeKafkaRecord_0_10.AUTO_OFFSET_RESET, ConsumeKafkaRecord_0_10.OFFSET_EARLIEST); runner.run(1, false); verify(mockConsumerPool, times(1)).obtainConsumer(anyObject(), 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); }
@OnStopped public void close() { final ConsumerPool pool = consumerPool; consumerPool = null; if (pool != null) { pool.close(); } }
ConsumeKafkaRecord_0_10 extends AbstractProcessor { @OnStopped public void close() { final ConsumerPool pool = consumerPool; consumerPool = null; if (pool != null) { pool.close(); } } }
ConsumeKafkaRecord_0_10 extends AbstractProcessor { @OnStopped public void close() { final ConsumerPool pool = consumerPool; consumerPool = null; if (pool != null) { pool.close(); } }  }
ConsumeKafkaRecord_0_10 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); }
ConsumeKafkaRecord_0_10 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);  }