MessageListenerThread extends Thread { <T> List<Message<T>> messages(List<ConsumerRecord<byte[], byte[]>> records, ActionLog actionLog, JSONReader<T> reader) throws IOException { int size = records.size(); actionLog.track("kafka", 0, size, 0); List<Message<T>> messages = new ArrayList<>(size); Set<String> correlationIds = new HashSet<>(); Set<String> clients = new HashSet<>(); Set<String> refIds = new HashSet<>(); Set<String> keys = Sets.newHashSetWithExpectedSize(size); long minTimestamp = Long.MAX_VALUE; for (ConsumerRecord<byte[], byte[]> record : records) { 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) correlationIds.add(correlationId); String client = header(headers, MessageHeaders.HEADER_CLIENT); if (client != null) clients.add(client); String refId = header(headers, MessageHeaders.HEADER_REF_ID); if (refId != null) refIds.add(refId); String key = key(record); keys.add(key); byte[] value = record.value(); long timestamp = record.timestamp(); logger.debug("[message] key={}, value={}, timestamp={}, refId={}, client={}, correlationId={}", key, new BytesLogParam(value), timestamp, refId, client, correlationId); if (minTimestamp > timestamp) minTimestamp = timestamp; T message = reader.fromJSON(value); messages.add(new Message<>(key, message)); } actionLog.context("key", keys.toArray()); if (!correlationIds.isEmpty()) actionLog.correlationIds = List.copyOf(correlationIds); if (!clients.isEmpty()) actionLog.clients = List.copyOf(clients); if (!refIds.isEmpty()) actionLog.refIds = List.copyOf(refIds); checkConsumerDelay(actionLog, minTimestamp, longConsumerDelayThresholdInNano); return messages; } MessageListenerThread(String name, MessageListener listener); @Override void run();  }
@Test void messages() throws IOException { ConsumerRecord<byte[], byte[]> record = new ConsumerRecord<>("topic", 0, 1, Strings.bytes("key"), Strings.bytes("{}")); record.headers().add(MessageHeaders.HEADER_CLIENT, Strings.bytes("client")); record.headers().add(MessageHeaders.HEADER_REF_ID, Strings.bytes("refId")); record.headers().add(MessageHeaders.HEADER_CORRELATION_ID, Strings.bytes("correlationId")); var actionLog = new ActionLog(null, null); List<Message<TestMessage>> messages = thread.messages(List.of(record), actionLog, JSONMapper.reader(TestMessage.class)); assertThat(messages).hasSize(1); assertThat(messages.get(0).key).isEqualTo("key"); assertThat(actionLog.context.get("key")).containsExactly("key"); assertThat(actionLog.clients).containsExactly("client"); assertThat(actionLog.refIds).containsExactly("refId"); assertThat(actionLog.correlationIds).containsExactly("correlationId"); }