private void runTask(final KafkaConsumer<String, Bytes> consumer) {
        while (true) {
            try {
                ConsumerRecords<String, Bytes> consumerRecords = consumer.poll(Duration.ofMillis(500L));
                if (!consumerRecords.isEmpty()) {
                    for (final ConsumerRecord<String, Bytes> record : consumerRecords) {
                        executor.submit(() -> Objects.requireNonNull(handlerMap.get(record.topic())).handle(record));
                    }
                    if (!enableKafkaMessageAutoCommit) {
                        consumer.commitAsync();
                    }
                }
            } catch (Exception e) {
                log.error("Kafka handle message error.", e);
            }
        }
    }
Configuring and initializing a KafkaConsumer client as a dispatcher to delivery Kafka Message to registered handler
by topic.
oap-server/server-fetcher-plugin/kafka-fetcher-plugin/src/main/java/org/apache/skywalking/oap/server/analyzer/agent/kafka/KafkaFetcherHandlerRegister.java