Kafka Connect 自定义 Connector 开发实战

public class MySourceConnector extends SourceConnector {
    private Map<String, String> configProps;

    @Override
    public void start(Map<String, String> props) {
        // 校验配置，尽早失败
        this.configProps = props;
        if (props.get(MySourceConfig.ENDPOINT) == null) {
            throw new ConnectException("endpoint 不能为空");
        }
    }

    @Override
    public Class<? extends Task> taskClass() {
        return MySourceTask.class;
    }

    @Override
    public List<Map<String, String>> taskConfigs(int maxTasks) {
        // 按外部系统的分片数切分，最多 maxTasks 个
        List<Map<String, String>> configs = new ArrayList<>();
        int shardCount = Math.min(maxTasks, queryShardCount());
        for (int i = 0; i < shardCount; i++) {
            Map<String, String> cfg = new HashMap<>(configProps);
            cfg.put(MySourceConfig.SHARD_ID, String.valueOf(i));
            configs.add(cfg);
        }
        return configs;
    }

    @Override
    public ConfigDef config() {
        return MySourceConfig.CONFIG_DEF;
    }

    @Override
    public String version() {
        return "1.0.0";
    }
}

public class MySourceTask extends SourceTask {
    private HttpClient client;
    private String shardId;
    private long lastOffset;

    @Override
    public void start(Map<String, String> props) {
        this.shardId = props.get(MySourceConfig.SHARD_ID);
        this.client = new HttpClient(props.get(MySourceConfig.ENDPOINT));
        // 从 offset 存储恢复断点，null 表示全新 Task
        Map<String, Object> offset = context.offsetStorageReader()
                .offset(Collections.singletonMap("shard", shardId));
        this.lastOffset = offset == null ? 0L : (Long) offset.get("position");
    }

    @Override
    public List<SourceRecord> poll() throws InterruptedException {
        List<Row> rows = client.fetch(shardId, lastOffset, BATCH_SIZE);
        if (rows.isEmpty()) {
            Thread.sleep(POLL_INTERVAL_MS);  // 空轮询必须退避
            return Collections.emptyList();
        }
        List<SourceRecord> records = new ArrayList<>(rows.size());
        for (Row row : rows) {
            Map<String, Object> partition = Collections.singletonMap("shard", shardId);
            Map<String, Object> offset = new HashMap<>();
            offset.put("position", row.position);
            records.add(new SourceRecord(partition, offset, topic,
                    Schema.STRING_SCHEMA, row.key, valueSchema, row.value));
            lastOffset = row.position;
        }
        return records;
    }
}

public class MySinkTask extends SinkTask {
    private Writer writer;
    private int remainingRetries;

    @Override
    public void start(Map<String, String> props) {
        this.writer = new Writer(props.get(MySinkConfig.ENDPOINT));
        this.remainingRetries = Integer.parseInt(
                props.getOrDefault(SinkConnectorConfig.MAX_RETRIES_CONFIG, "10"));
    }

    @Override
    public void put(Collection<SinkRecord> records) {
        if (records.isEmpty()) {
            return;
        }
        for (SinkRecord record : records) {
            writer.stage(record.topic(), record.kafkaPartition(),
                         record.kafkaOffset(), record.value());
        }
        // 整批一次性提交，失败整批重试
        retry(() -> writer.commitBatch());
    }

    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> currentOffsets) {
        // Worker 在提交 offset 前调用，确保数据已真正落外部系统
        writer.flush();
    }
}

{
  "errors.tolerance": "all",
  "errors.log.enable": true,
  "errors.log.include.messages": true,
  "errors.deadletterqueue.topic.name": "dlq-my-sink",
  "errors.deadletterqueue.context.headers.enable": true
}

// 幂等 upsert：重复批次不会产生重复行
// INSERT INTO t (topic, partition, offset, payload)
// VALUES (?, ?, ?, ?)
// ON CONFLICT (topic, partition, offset) DO NOTHING;

// 外部系统支持事务时，整批写入与位点记录同一事务
writer.beginTransaction();
try {
    for (SinkRecord r : batch) { writer.write(r); }
    writer.recordOffsets(batch);   // 把位点与数据放同一事务
    writer.commit();
} catch (Exception e) {
    writer.rollback();
    throw e;
}

public class MaskFieldTransform<R extends ConnectRecord<R>> implements Transformation<R> {
    private List<String> fields;

    @Override
    public void configure(Map<String, ?> configs) {
        this.fields = Arrays.asList(
                String.valueOf(configs.get("fields")).split(","));
    }

    @Override
    public R apply(R record) {
        if (record.value() == null) {
            return record;
        }
        Struct value = (Struct) record.value();
        for (String f : fields) {
            if (value.schema().field(f) != null) {
                value.put(f, "***");
            }
        }
        return record.newRecord(record.topic(), record.kafkaPartition(),
                record.keySchema(), record.key(),
                value.schema(), value, record.timestamp());
    }

    @Override
    public ConfigDef config() {
        return new ConfigDef().define("fields", ConfigDef.Type.STRING,
                ConfigDef.Importance.HIGH, "需要脱敏的字段");
    }
}

{
  "transforms": "maskPwd,route",
  "transforms.maskPwd.type": "com.example.MaskFieldTransform",
  "transforms.maskPwd.fields": "password,token",
  "transforms.route.type": "org.apache.kafka.connect.transforms.RegexRouter",
  "transforms.route.regex": "orders-(.*)",
  "transforms.route.replacement": "orders-clean-$1",
  "predicates": "isOrder",
  "predicates.isOrder.type": "org.apache.kafka.connect.transforms.predicates.TopicNameMatches",
  "predicates.isOrder.pattern": "orders-.*",
  "transforms.maskPwd.predicate": "isOrder"
}

# 创建 Connector
curl -X POST http://connect:8083/connectors \
  -H "Content-Type: application/json" \
  -d @my-connector.json

# 查看状态与 Task 分布
curl http://connect:8083/connectors/my-connector/status

# 暂停 / 恢复 / 删除
curl -X PUT  http://connect:8083/connectors/my-connector/pause
curl -X DELETE http://connect:8083/connectors/my-connector

# 更新配置（注意：会触发 Task 重启）
curl -X PUT http://connect:8083/connectors/my-connector/config \
  -H "Content-Type: application/json" -d @new-config.json