如何测量Apache Kafka数据写入MongoDB的端到端耗时？

我使用以下docker-compose文件搭建Apache Kafka与MongoDB的连接环境：

version: "3.6"
services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.2.2
    hostname: zookeeper
    container_name: zookeeper
    networks:
      - localnet
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      KAFKA_JMX_PORT: 35000

  broker:
    image: confluentinc/cp-kafka:7.2.2
    hostname: broker
    container_name: broker
    ports:
      - "9092:9092"
    depends_on:
      - zookeeper
    networks:
      - localnet
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181"
      KAFKA_LISTENERS: LISTENER_1://broker:29092,LISTENER_2://broker:9092
      KAFKA_ADVERTISED_LISTENERS: LISTENER_1://broker:29092,LISTENER_2://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: LISTENER_1:PLAINTEXT,LISTENER_2:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: LISTENER_1
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1
      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0
      CONFLUENT_SUPPORT_CUSTOMER_ID: "anonymous"
      KAFKA_DELETE_TOPIC_ENABLE: "true"
      KAFKA_JMX_PORT: 35000
      
  schema-registry:
    image: confluentinc/cp-schema-registry:7.2.2
    hostname: schema-registry
    container_name: schema-registry
    depends_on:
      - broker
    networks:
      - localnet
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: "broker:29092"
      SCHEMA_REGISTRY_LISTENERS: http://0.0.0.0:8081    
          
  connect:
    build:
      context: .
      dockerfile: connect.Dockerfile
    ports:
      - "35000:35000"
      - "8083:8083"
    hostname: connect
    container_name: connect
    depends_on:
      - zookeeper
      - broker
    networks:
      - localnet
    environment:
      KAFKA_JMX_PORT: 35000
      KAFKA_JMX_HOSTNAME: localhost
      CONNECT_BOOTSTRAP_SERVERS: "broker:29092"
      CONNECT_REST_ADVERTISED_HOST_NAME: connect
      CONNECT_GROUP_ID: connect-cluster-group
      CONNECT_CONFIG_STORAGE_TOPIC: docker-connect-configs
      CONNECT_CONFIG_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_OFFSET_FLUSH_INTERVAL_MS: 10000
      CONNECT_OFFSET_STORAGE_TOPIC: docker-connect-offsets
      CONNECT_OFFSET_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_STATUS_STORAGE_TOPIC: docker-connect-status
      CONNECT_STATUS_STORAGE_REPLICATION_FACTOR: 1
      CONNECT_ZOOKEEPER_CONNECT: "zookeeper:2181"
      CONNECT_PLUGIN_PATH: "/usr/share/java,/usr/share/confluent-hub-components"
      CONNECT_CONNECTIONS_MAX_IDLE_MS: 180000
      CONNECT_METADATA_MAX_AGE_MS: 180000
      CONNECT_AUTO_CREATE_TOPICS_ENABLE: "true"
      CONNECT_KEY_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"
      CONNECT_VALUE_CONVERTER: "org.apache.kafka.connect.json.JsonConverter"

  mongo1:
    image: mongo:6.0.5
    container_name: mongo1
    ports:
      - "35001:27017"
    depends_on:
      - zookeeper
      - broker
      - connect
    networks:
      - localnet
    restart: always

  mongo1-setup:
    image: mongo:6.0.5
    container_name: mongo1-setup
    volumes: 
      - ./config-replica.js:/config-replica.js
    depends_on:
      - mongo1
    networks:
      - localnet
    entrypoint:
      [
        "bash",
        "-c",
        "sleep 10 && mongosh --host mongo1:27017 config-replica.js && sleep 10",
      ]
    restart: "no"
    
    
  mongo-express:
    image: mongo-express
    restart: always
    ports:
      - 8081:8081
    environment:
      ME_CONFIG_MONGODB_SERVER: mongo1
      ME_CONFIG_MONGODB_PORT: 27017
      ME_CONFIG_OPTIONS_EDITORTHEME: "default"
    networks:
      - localnet
    depends_on:
      - mongo1 

networks:
  localnet:
    attachable: true

对应的connect.Dockerfile内容如下：

FROM confluentinc/cp-kafka-connect:7.2.5

RUN confluent-hub install --no-prompt --verbose mongodb/kafka-connect-mongodb:latest

ENV CONNECT_PLUGIN_PATH="/usr/share/java,/usr/share/confluent-hub-components"

我希望测量数据从写入Apache Kafka到保存至MongoDB的时间，请问最简测量方法是什么？我曾尝试使用JMX，但未找到可用于测量该耗时的指标。

方案一：消息自带时间戳+Mongo自动入库时间 #

这是最直接的方案，无需依赖日志或监控工具：

• 生产消息时，在消息体中加入发送时间戳（毫秒级，比如用System.currentTimeMillis()生成），示例消息结构：

  {
    "kafka_send_ts": 1718000000000,
    "content": "your_business_data"
  }
  

• 配置MongoDB集合，让文档插入时自动添加入库时间戳：
  进入Mongo容器执行命令：

  // 给目标集合添加默认值为当前时间的字段
  db.your_collection.createIndex({mongo_insert_ts: 1})
  db.runCommand({
    collMod: "your_collection",
    "validator": {},
    "validationLevel": "off",
    "validationAction": "warn",
    "default": {mongo_insert_ts: new Date()}
  })
  

• 后续从Mongo中取出消息，用mongo_insert_ts的时间戳减去kafka_send_ts，即可得到单条消息的端到端耗时，批量测试后取平均值更准确。

方案二：日志对比法 #

适合不想修改消息结构的快速验证：

• 开启Kafka Connect的DEBUG日志：在Connect的环境变量中添加LOG4J_LOGGERS=com.mongodb.kafka.connect=DEBUG，重启Connect后，日志会打印每条消息的处理时间点。
• 开启MongoDB的全量写操作日志：进入Mongo容器执行db.setProfilingLevel(2)，该命令会记录所有数据库的写操作及对应的时间。
• 找到同一条消息在Connect日志中的处理开始时间，和Mongo日志中的写入完成时间，两者的差值即为端到端耗时。

补充说明 #

MongoDB Kafka Connect Sink本身没有暴露端到端耗时的JMX指标，所以你之前找不到是正常的，上述两种方案都是无需额外工具的最简实现方式。

内容的提问来源于stack exchange，提问作者Kamivix