如何通过Callback使Kafka Producer的send()返回RecordMetadata或null

这个问题其实戳中了Kafka Producer异步模型的核心点——send()方法本身是异步非阻塞的，它没办法直接返回回调里的RecordMetadata，因为回调逻辑是在后台I/O线程中执行的，而send()调用后会立刻返回。另外你当前的写法还有个小问题：Callback接口的onCompletion方法是void返回类型的，你写的带返回值的方法其实不符合接口规范，编译器会直接报错。

下面给你两种可行的解决方案，兼顾同步/异步场景：

方案1：同步等待Future结果（最简单直接） #

Kafka Producer的send()方法本身会返回一个Future<RecordMetadata>，你可以直接调用Future.get()来同步等待发送完成，这样就能拿到RecordMetadata，同时统一处理异常：

try {
    // 同步等待发送结果，拿到RecordMetadata
    RecordMetadata recordMetadata = kafkaProducer.send(
        new ProducerRecord<>(topic, null, timestamp, key, message)
    ).get();
    // 这里可以直接使用recordMetadata做后续业务处理
} catch (InterruptedException e) {
    // 处理线程中断异常
    Thread.currentThread().interrupt();
} catch (ExecutionException e) {
    // 处理发送失败的根因（比如网络异常、主题不存在等）
    Exception sendException = (Exception) e.getCause();
    // 对应你之前逻辑里"返回null"的场景，这里可以做异常兜底处理
}

这种方式的优势是代码简洁，适合必须等待发送结果才能继续的同步场景；缺点是会阻塞当前线程，直到发送完成或失败。

方案2：用CompletableFuture封装回调（兼顾异步灵活性） #

如果你想保留回调的异步处理特性，同时让调用方能够灵活获取结果，可以用CompletableFuture来封装回调逻辑，把结果传递到Future中：

public CompletableFuture<RecordMetadata> sendWithCallback(String topic, Long timestamp, String key, String message) {
    CompletableFuture<RecordMetadata> resultFuture = new CompletableFuture<>();
    
    kafkaProducer.send(
        new ProducerRecord<>(topic, null, timestamp, key, message),
        (metadata, exception) -> {
            if (exception == null) {
                // 发送成功，把metadata传递给Future
                resultFuture.complete(metadata);
            } else {
                // 发送失败，传递异常（也可以用resultFuture.complete(null)来返回null，更推荐传递异常让调用方处理）
                resultFuture.completeExceptionally(exception);
            }
        }
    );
    
    return resultFuture;
}

调用方可以根据需求选择异步或同步处理：

// 异步处理方式
sendWithCallback(topic, timestamp, key, message)
    .thenAccept(metadata -> {
        // 拿到metadata后的业务逻辑
    })
    .exceptionally(exception -> {
        // 处理发送失败的情况，对应你之前返回null的逻辑
        return null;
    });

// 同步等待方式
try {
    RecordMetadata metadata = sendWithCallback(topic, timestamp, key, message).get();
} catch (Exception e) {
    // 统一处理异常
}

这种方式既保留了Kafka Producer的异步性能，又能让调用方灵活控制结果的处理方式，比原生回调更易扩展。

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