MessageListenerThread extends Thread { <T> void handleBulk(String topic, MessageProcess<T> process, List<ConsumerRecord<byte[], byte[]>> records, double longProcessThresholdInNano) { ActionLog actionLog = logManager.begin("=== message handling begin ===", null); try { actionLog.action("topic:" + topic); actionLog.context("topic", topic); actionLog.context("handler", process.bulkHandler.getClass().getCanonicalName()); List<Message<T>> messages = messages(records, actionLog, process.reader); for (Message<T> message : messages) { process.validator.validate(message.value, false); } process.bulkHandler.handle(messages); } catch (Throwable e) { logManager.logError(e); } finally { long elapsed = actionLog.elapsed(); checkSlowProcess(elapsed, longProcessThresholdInNano); logManager.end("=== message handling end ==="); } } MessageListenerThread(String name, MessageListener listener); @Override void run();  }
@Test void handleBulk() throws Exception { var key = "key"; var message = new TestMessage(); message.stringField = "value"; var record = new ConsumerRecord<>("topic", 0, 0, System.currentTimeMillis(), TimestampType.CREATE_TIME, -1, -1, -1, Strings.bytes(key), Strings.bytes(JSON.toJSON(message))); record.headers().add(MessageHeaders.HEADER_CORRELATION_ID, Strings.bytes("correlationId")); record.headers().add(MessageHeaders.HEADER_REF_ID, Strings.bytes("refId")); thread.handleBulk("topic", new MessageProcess<>(null, bulkMessageHandler, TestMessage.class), List.of(record), Duration.ofHours(1).toNanos()); verify(bulkMessageHandler).handle(argThat(value -> value.size() == 1 && key.equals(value.get(0).key) && "value".equals(value.get(0).value.stringField))); }