Kafka源连接器中只有一次语义

import org.apache.kafka.connect.source.SourceTask;
import org.apache.kafka.connect.source.SourceTaskContext;
import org.apache.kafka.connect.storage.OffsetStorageReader;

public class MySourceTask extends SourceTask {
    @Override
    public void start(Map<String, String> props) {
        // 初始化连接器
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        // 从外部系统读取数据
        List<SourceRecord> records = readRecords();
        
        // 使用事务API提交偏移量
        try {
            Map<TopicPartition, OffsetAndMetadata> offsets = new HashMap<>();
            for (SourceRecord record : records) {
                TopicPartition tp = new TopicPartition(record.topic(), record.partition());
                offsets.put(tp, new OffsetAndMetadata(record.offset()));
            }
            context.offset(offsets);
        } catch (Exception e) {
            // 处理异常
        }
        
        return records;
    }

    @Override
    public void commit() throws InterruptedException {
        // 提交事务
    }

    @Override
    public void stop() {
        // 停止连接器
    }
}