如何从Kafka集群采集Producer配置？需获取client.id及发布主题

Great question! Since producer configurations are stored locally on the client side, the Kafka cluster itself doesn’t maintain a centralized repository of these details. But there are several practical approaches to collect the client.id and the set of topics a producer is publishing to:

1. Leverage Kafka's Built-in Metrics #

Kafka producers expose JMX (or Prometheus-compatible) metrics that include both client.id and topic-specific metadata. For example, metrics under the kafka.producer:type=producer-metrics namespace will be tagged with client.id and topic labels for each topic the producer writes to:

• Use tools like JConsole to connect directly to the producer's JVM and browse these metrics.
• If you use Prometheus, deploy the official Kafka Prometheus Exporter to scrape these metrics. Look for metrics like record_send_rate_total or records_sent_total—each will have the relevant client_id and topic tags to map producers to their target topics.

2. Parse Producer Client Logs #

If you have access to the producer client's log files, you can extract the required information by parsing them:

• client.id: Producers log their full config on startup. Look for lines starting with ProducerConfig values:—you’ll find client.id = <your-client-id> in that section.
• Topics: When the producer sends records, logs will include entries like Sending record to topic <topic-name> partition.... Use tools like grep, awk, or a log aggregation stack (ELK, Grafana Loki) to filter and aggregate this data to get the full set of topics per producer.

3. Implement a Custom Producer Interceptor #

If you control the producer codebase, adding a ProducerInterceptor is a reliable way to capture this data in real-time:

• The interceptor can access the producer’s config (including client.id) during initialization, and intercept every record being sent to capture its topic.
• Here’s a simple Java example:

  public class ProducerConfigTracker implements ProducerInterceptor<String, Object> {
      private String clientId;
  
      @Override
      public void configure(Map<String, ?> configs) {
          // Extract client.id from the producer config
          this.clientId = (String) configs.get(ProducerConfig.CLIENT_ID_CONFIG);
      }
  
      @Override
      public ProducerRecord<String, Object> onSend(ProducerRecord<String, Object> record) {
          // Capture clientId and target topic—you can log this or send to a monitoring system
          System.out.printf("Producer [%s] is publishing to topic: %s%n", clientId, record.topic());
          return record;
      }
  
      @Override
      public void onAcknowledgement(RecordMetadata metadata, Exception exception) {
          // Optional: Handle acknowledgement logic if needed
      }
  
      @Override
      public void close() {
          // Cleanup resources if needed
      }
  }
  

• To enable the interceptor, add this line to your producer config:

  props.put(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG, "com.yourorg.ProducerConfigTracker");
  

4. Use Broker Logs (Limited Use Case) #

Broker logs can provide partial visibility, though it’s less precise than the above methods:

• When a producer connects to the broker, logs will include entries like Connection from /192.168.1.100 with clientId <client-id>.
• Broker logs also record incoming records with their target topic, but you’ll need to correlate the client’s IP/connection ID with the topic entries to map client.id to topics. This works best if you have isolated producers (one per IP), but is unreliable for shared environments.

内容的提问来源于stack exchange，提问作者Ihor M.