Kafka Connect Source Connector内存溢出问题求助

我之前开发文件读取类的Kafka Connect Source插件时，也踩过一模一样的OOM坑！结合Kafka Connect的工作机制和实际调优经验，给你几个切实可行的解决方案：

1. 严格限制每次poll()返回的记录批次大小 #

这是解决OOM最核心的手段——你之前应该是一次性把整个大文件的所有行都转换成SourceRecord塞进列表返回，直接撑爆内存。Kafka Connect本身是支持批次处理的，我们要做的就是控制每次poll()只返回固定数量的记录：

• 新增一个可配置的批次大小参数（比如batch.size，默认设为500或1000，根据你的内存情况调整）
• 在poll()方法里，读取到指定数量的记录就停止，返回当前批次，下一次poll()再继续读取后续内容

示例代码：

private int batchSize = 500; // 可通过配置参数注入
private int currentLineNumber = 0;
private LineNumberReader lineReader;

@Override
public List<SourceRecord> poll() throws InterruptedException {
    List<SourceRecord> records = new ArrayList<>(batchSize);
    String line;
    try {
        // 循环读取，直到达到批次大小或文件结束
        while ((line = lineReader.readLine()) != null && records.size() < batchSize) {
            // 绑定预缓存的Schema创建SourceRecord
            SourceRecord record = buildSourceRecord(line, currentLineNumber);
            records.add(record);
            currentLineNumber++;
        }
        // 文件读取完毕的标记
        if (line == null) {
            this.isFinished = true;
        }
    } catch (IOException e) {
        throw new RuntimeException("Failed to read file line", e);
    }
    return records;
}

2. 缓存Avro Schema，避免重复创建对象 #

如果你的代码里每次创建SourceRecord都重新生成Avro Schema，会产生大量重复的Schema对象，严重占用内存。把Schema作为类的成员变量初始化一次即可：

// 类级别的Schema缓存，只初始化一次
private final Schema fileRecordSchema;

public CustomFileSourceTask() {
    this.fileRecordSchema = SchemaBuilder.record("FileLineRecord")
        .fields()
        .name("content").type(Schema.STRING_SCHEMA).noDefault()
        .name("lineNumber").type(Schema.INT_SCHEMA).noDefault()
        .build();
}

// 创建SourceRecord时直接复用缓存的Schema
private SourceRecord buildSourceRecord(String line, int lineNum) {
    GenericRecord avroRecord = new GenericData.Record(fileRecordSchema);
    avroRecord.put("content", line);
    avroRecord.put("lineNumber", lineNum);
    
    return new SourceRecord(
        getSourcePartition(), // 你的源分区定义
        Collections.singletonMap("lineNumber", lineNum), // 偏移量
        targetTopic, // 目标Topic
        fileRecordSchema,
        avroRecord
    );
}

3. 配合commit()正确管理偏移量，避免重复读取 #

commit()方法的核心作用是把当前处理到的偏移量（比如行号）持久化到Kafka Connect的offset存储中。这样即使插件重启，也能从上次中断的位置继续读取，不用从头加载整个文件，同时也能配合批次处理，确保每一批记录处理完成后再提交偏移量：

@Override
public void commit() throws InterruptedException {
    // 构建当前偏移量
    Map<String, Object> currentOffset = Collections.singletonMap("lineNumber", currentLineNumber);
    // 写入偏移量并刷盘
    context.offsetStorageWriter().offsets(Map.of(getSourcePartition(), currentOffset));
    context.offsetStorageWriter().flush();
}

注意：Kafka Connect会在合适的时机调用commit()（比如当一批记录被成功发送到Kafka后），我们只需要在这个方法里正确保存偏移量即可。

4. 改用流式读取替代LineNumberReader（进阶优化） #

如果文件特别大，LineNumberReader可能还是会在底层缓存较多内容，改用Java NIO的流式读取可以进一步降低内存占用——Files.lines()是懒加载的，只会在需要时读取下一行：

private Stream<String> fileLineStream;
private Iterator<String> lineIterator;

@Override
public void start(Map<String, String> props) {
    Path filePath = Paths.get(props.get("file.path"));
    try {
        fileLineStream = Files.lines(filePath, StandardCharsets.UTF_8);
        lineIterator = fileLineStream.iterator();
        
        // 从已保存的偏移量恢复读取位置
        Map<String, Object> savedOffset = context.offsetStorageReader().offset(getSourcePartition());
        if (savedOffset != null) {
            int savedLineNum = (Integer) savedOffset.get("lineNumber");
            // 跳过已经处理过的行
            for (int i = 0; i < savedLineNum && lineIterator.hasNext(); i++) {
                lineIterator.next();
            }
            currentLineNumber = savedLineNum;
        }
    } catch (IOException e) {
        throw new RuntimeException("Failed to open target file", e);
    }
}

@Override
public List<SourceRecord> poll() throws InterruptedException {
    List<SourceRecord> records = new ArrayList<>(batchSize);
    while (lineIterator.hasNext() && records.size() < batchSize) {
        String line = lineIterator.next();
        records.add(buildSourceRecord(line, currentLineNumber));
        currentLineNumber++;
    }
    // 文件读取完毕后关闭流
    if (!lineIterator.hasNext()) {
        this.isFinished = true;
        try {
            fileLineStream.close();
        } catch (IOException e) {
            log.error("Error closing file stream", e);
        }
    }
    return records;
}

总结 #

这几个方案组合起来基本就能解决大文件读取的OOM问题：

• 核心是分批次返回记录，避免一次性加载全部数据到内存
• 辅助优化包括缓存Schema、正确管理偏移量、改用流式读取

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