如何在调用REST API的自定义Kafka Connect Sink中处理背压

嘿，这个问题刚好戳中了Kafka Connect Sink设计的核心矛盾——既要遵守框架“不能在put()里阻塞”的规则，又要应对下游系统处理慢的情况。我来分享几个实际项目里验证过的方案，帮你搞定背压和OOM的问题：

1. 先从配置层面掐住数据流入的源头 #

Kafka Connect本身就自带了不少流量控制参数，先把这些调好能省不少事：

• consumer.max.poll.records：限制Consumer每次从Broker拉取的记录数。别贪多，根据你的REST API吞吐能力来设——比如API每秒能处理100条，就把这个值设成100-200，避免一次性拉太多数据把缓存撑爆。
• batch.size（Sink专属配置）：定义每次flush()时要处理的记录量。这个值要和API的处理能力匹配，让flush()的频率刚好跟上API的速度，别让数据在缓存里堆太久。
• connect.max.retries + retry.backoff.ms：当API调用超时或失败时，让Sink自动重试，同时通过退避时间给下游系统留缓冲，避免死磕着压测。

2. 实现有界缓存+主动触发flush，用异常传递背压 #

自定义SinkTask里绝对不能用无界容器存数据（比如ArrayList），必须整个有界的队列（比如ArrayBlockingQueue），然后这么玩：

• 在put()里，用非阻塞的方式往队列加数据（比如offer()方法）。如果队列满了，直接抛出RetriableException——这是Connect内置的重试异常，框架看到这个异常会暂停当前任务的调度，过会儿再重试put()，相当于给上游发了"我忙不过来，先别给数据"的信号，完美传递背压。
• 另外，在put()里检查队列大小，一旦达到batch.size的阈值，主动调用flush()（注意：真正的阻塞逻辑要放在框架调用的flush()方法里，这里只是触发处理）。

给你个简化的代码示例参考：

public class RestApiSinkTask extends SinkTask {
    private BlockingQueue<SinkRecord> buffer;
    private int batchSize;
    private RestApiClient apiClient;

    @Override
    public void start(Map<String, String> props) {
        batchSize = Integer.parseInt(props.getOrDefault("batch.size", "100"));
        // 队列设为批量的2倍，留够缓冲空间，同时绝对不允许无界增长
        buffer = new ArrayBlockingQueue<>(batchSize * 2);
        apiClient = new RestApiClient(props.get("api.url"));
    }

    @Override
    public void put(Collection<SinkRecord> records) {
        for (SinkRecord record : records) {
            // 非阻塞添加，失败就抛重试异常触发背压
            if (!buffer.offer(record)) {
                throw new RetriableException("Buffer is full - backpressure activated");
            }
        }
        // 达到批量阈值，主动触发flush
        if (buffer.size() >= batchSize) {
            flush(null);
        }
    }

    @Override
    public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
        List<SinkRecord> batch = new ArrayList<>(batchSize);
        // 从队列里取出一批数据
        buffer.drainTo(batch, batchSize);
        if (batch.isEmpty()) return;

        // 这里放心阻塞调用API，Connect允许在flush()里做同步等待
        try {
            apiClient.sendBatch(batch);
            // 成功发送后提交偏移量
            if (offsets != null) {
                context.offset(offsets);
            }
        } catch (ApiTimeoutException | IOException e) {
            // 发送失败，把数据放回队列头部，等待下次重试
            buffer.addAll(0, batch);
            throw new RetriableException("API call failed, will retry", e);
        }
    }
}

3. 避免OOM的核心：卡死缓存的上限 #

上面代码里把队列大小设为batch.size * 2，这个值你可以根据自己的内存情况调整，但绝对不能设成无界。一旦队列满了，就通过RetriableException让框架暂停推送数据，从根本上杜绝缓存无限增长导致的OOM。

4. 搞懂Connect的任务调度逻辑，为啥flush()阻塞是安全的 #

Kafka Connect的SinkTask是由单个线程调度的：这个线程先调用put()收数据，然后到点调用flush()处理。如果flush()因为API慢而阻塞，这个线程就会被占着，没法继续调用put()，上游的Consumer自然也就不会再拉新的记录——这就是Connect内置的背压传递链条，完全符合官方文档的要求。

总结一下：用有界缓存限制内存占用，用重试异常触发上游背压，用flush()阻塞等待API响应，再配合合理的批量配置，就能完美解决下游慢带来的问题。

内容的提问来源于stack exchange，提问作者Chris W.