@Override
  public void run() {
    try (KafkaConsumer<byte[], byte[]> kafkaConsumer = new KafkaConsumer<>(properties)) {
      kafkaConsumer.subscribe(
        topics,
        // added for integration tests only, see ITKafkaCollector
        new ConsumerRebalanceListener() {
          @Override
          public void onPartitionsRevoked(Collection<TopicPartition> partitions) {
            // technically we should remove only the revoked partitions but for test purposes it
            // does not matter
            assignedPartitions.set(List.of());
          }

          @Override
          public void onPartitionsAssigned(Collection<TopicPartition> partitions) {
            assignedPartitions.set(List.copyOf(partitions));
          }
        });
      LOG.debug("Kafka consumer starting polling loop.");
      while (running.get()) {
        final ConsumerRecords<byte[], byte[]> consumerRecords = kafkaConsumer.poll(Duration.of(1000, ChronoUnit.MILLIS));
        LOG.debug("Kafka polling returned batch of {} messages.", consumerRecords.count());
        for (ConsumerRecord<byte[], byte[]> record : consumerRecords) {
          final byte[] bytes = record.value();
          metrics.incrementMessages();
          metrics.incrementBytes(bytes.length);

          if (bytes.length == 0) continue; // lenient on empty messages

          if (bytes.length < 2) { // need two bytes to check if protobuf
            metrics.incrementMessagesDropped();
          } else {
            // If we received legacy single-span encoding, decode it into a singleton list
            if (!protobuf3(bytes) && bytes[0] <= 16 && bytes[0] != 12 /* thrift, but not list */) {
              Span span;
              try {
                span = SpanBytesDecoder.THRIFT.decodeOne(bytes);
              } catch (RuntimeException e) {
                metrics.incrementMessagesDropped();
                continue;
              }
              collector.accept(List.of(span), NOOP);
            } else {
              collector.acceptSpans(bytes, NOOP);
            }
          }
        }
      }
    } catch (RuntimeException | Error e) {
      LOG.warn("Unexpected error in polling loop spans", e);
      throw e;
    } finally {
      LOG.debug("Kafka consumer polling loop stopped. Kafka consumer closed.");
    }
  }
Consumes spans from Kafka messages, ignoring malformed input
zipkin-collector/kafka/src/main/java/zipkin2/collector/kafka/KafkaCollectorWorker.java