KafkaMessageListenerContainer implements Runnable, AutoCloseable { public void init() { executorService.submit(this); } KafkaMessageListenerContainer(Properties kafkaConfig,
                                         OffsetStore offsetStore,
                                         String topic,
                                         BiConsumer<K, V> messageListener); void setExceptionHandler(BiConsumer<Tuple2<K, V>, Exception> exceptionHandler); void init(); @Override void run(); @Override void close(); long getRecordsProcessed(); static final int DEFAULT_POLL_TIMEOUT; }
@Test public void testReceive_errorHandling() throws Exception { int messagesToSend = 1000; preloadTopic(TOPIC, messagesToSend); Properties configs = props(getConsumerConfigs()); AtomicInteger messagesReceived = new AtomicInteger(); AtomicInteger exceptionsHandled = new AtomicInteger(); CountDownLatch latch = new CountDownLatch(messagesToSend); MemoryOffsetStore offsetStore = new MemoryOffsetStore(); KafkaMessageListenerContainer.Builder<byte[], byte[]> builder = new KafkaMessageListenerContainer.Builder<byte[], byte[]>() .kafkaConfig(configs) .offsetStore(offsetStore) .topic(TOPIC) .messageListener((key, value) -> { try { if (messagesReceived.incrementAndGet() == 500) { throw new IllegalArgumentException("Boom!"); } } finally { latch.countDown(); } }) .exceptionHandler((tuple, exception) -> { exceptionsHandled.incrementAndGet(); assertTrue(exception instanceof IllegalArgumentException); }); try (KafkaMessageListenerContainer<byte[], byte[]> container = builder.build()) { container.init(); if (!latch.await(10, TimeUnit.SECONDS)) { fail("Timeout expired waiting on latch"); } assertEquals(messagesToSend, messagesReceived.get()); assertEquals(1, exceptionsHandled.get()); } }