MessageListenerThread extends Thread { <T> void handle(String topic, MessageProcess<T> process, List<ConsumerRecord<byte[], byte[]>> records, double longProcessThresholdInNano) { for (ConsumerRecord<byte[], byte[]> record : records) { ActionLog actionLog = logManager.begin("=== message handling begin ===", null); try { actionLog.action("topic:" + topic); actionLog.context("topic", topic); actionLog.context("handler", process.handler.getClass().getCanonicalName()); actionLog.track("kafka", 0, 1, 0); Headers headers = record.headers(); if ("true".equals(header(headers, MessageHeaders.HEADER_TRACE))) actionLog.trace = true; String correlationId = header(headers, MessageHeaders.HEADER_CORRELATION_ID); if (correlationId != null) actionLog.correlationIds = List.of(correlationId); String client = header(headers, MessageHeaders.HEADER_CLIENT); if (client != null) actionLog.clients = List.of(client); String refId = header(headers, MessageHeaders.HEADER_REF_ID); if (refId != null) actionLog.refIds = List.of(refId); logger.debug("[header] refId={}, client={}, correlationId={}", refId, client, correlationId); String key = key(record); actionLog.context("key", key); long timestamp = record.timestamp(); logger.debug("[message] timestamp={}", timestamp); checkConsumerDelay(actionLog, timestamp, longConsumerDelayThresholdInNano); byte[] value = record.value(); logger.debug("[message] value={}", new BytesLogParam(value)); T message = process.reader.fromJSON(value); process.validator.validate(message, false); process.handler.handle(key, message); } 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 handle() 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_TRACE, Strings.bytes("true")); record.headers().add(MessageHeaders.HEADER_CLIENT, Strings.bytes("client")); thread.handle("topic", new MessageProcess<>(messageHandler, null, TestMessage.class), List.of(record), Duration.ofHours(1).toNanos()); verify(messageHandler).handle(eq(key), argThat(value -> "value".equals(value.stringField))); }