Kafka Connect SinkTask实例隔离性与Java对象状态管理

package org.MySink.influxSink;

import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.sink.SinkTask;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collection;
import java.util.Map;

public class MySinkTask extends SinkTask {
  // 静态变量，Logger实例通常是线程安全的，所有任务实例共享同一个Logger对象
  private static Logger log = LoggerFactory.getLogger(MySinkTask.class); 

  private String influxMeasurement; // 实例变量
  private MySinkConnectorConfig config; // 实例变量
  private Map<String, String> configMap; // 实例变量

  @Override
  public String version() {
    return VersionUtil.getVersion();
  }

  @Override
  public void start(Map<String, String> map) {
    // 每个任务实例在启动时，都会接收到其专属的配置map
    config = new MySinkConnectorConfig(map);
    configMap = map;

    // influxMeasurement 是当前任务实例特有的变量，根据传入的配置进行初始化
    influxMeasurement = config.getInfluxMeasurement(); 
  }

  @Override
  public void put(Collection<SinkRecord> collection) {
      if(collection.isEmpty()) {
          return;
      }

      final SinkRecord first = collection.iterator().next();
      final int recordsCount = collection.size();

      // 这里的 influxMeasurement 应该始终引用当前任务实例的私有值
      log.info(influxMeasurement + ": Received {} records. First record Kafka coordinates: ({}-{}-{}).",
              recordsCount, first.topic(), first.kafkaPartition(), first.kafkaOffset());
  }

  @Override
  public void flush(Map<TopicPartition, OffsetAndMetadata> map) {
    // 资源刷新逻辑，例如将缓冲区数据写入目标系统
  }

  @Override
  public void stop() {
    // 关闭资源，清理状态
  }
}