基于Kafka封装异步发送库：CompletableFuture回调相关问题确认

我们基于旧版Kafka封装轻量级工具库，kafkaTemplate.send(...)不支持.whenComplete()方法，初始实现如下：

1. RetryTemplate Bean定义 #

@Bean
public RetryTemplate prodRetryTemplate() {    
   final RetryTemplate template = RetryTemplate.builder().maxAttempts(3)
         .retryOn(TopicAuthorizationException.class).exponentialBackoff(1000, 2, 2000).build();
   return template;
}

2. 异步发送方法实现 #

@Override
@Async(EXECUTOR_SERVICE) //已定义Executor Bean，此处省略
public CompletableFuture<DtoResponse> asyncSend(final Integer key, final String topic, final String message)
      throws Exception {

   final ProducerRecord<K, V> producerRecord = buildRecord(key, topic, message);

   DtoResponse[] responseHolder = new DtoResponse[1];
   template.execute(ctx ->{

            final ListenableFuture<SendResult<Integer, String>> future = kafkaTemplate.send(producerRecord);
            future.addCallback(new ListenableFutureCallback<SendResult<Integer, String>>() {

               @Override
               public void onSuccess(final SendResult<Integer, String> result) {
                  log.info("success...");
                  responseHolder[0] = DtoResponse.builder()
                        .error(null)
                        .success(true)
                        .msg("PUBLISHED")
                        .build();
               }

               @SneakyThrows
               @Override
               public void onFailure(final Throwable ex) {
                  log.error("error...");
                  throwError(producerRecord, eventName, ex); //执行额外日志记录
               }
            });

      return null; //适配RetryTemplate的execute方法返回要求
   });

   return CompletableFuture.completedFuture(responseHolder[0]);
}

调用asyncSend时，是否可能返回null？也就是return CompletableFuture.completedFuture(null);会不会在onSuccess或onFailure执行前就执行？

客户端调用示例：

public void myService(){  
  MyLib lib = new MyLib();
  CompletableFuture<DtoResponse> future = lib.sendAsync(key, topic, message);
  future.whenComplete((result, throwable) -> { 
     //是否可能result和throwable同时为null？
    });
}

Update 1 #

调整实现后，客户端需要在whenComplete中处理SendResult和异常，但我们希望在工具库内部完成onSuccess/onFailure的关键日志，同时让客户端明确消息发送状态，该怎么处理？

Update 2 #

调整后发现，当触发Kafka构造异常（比如把oauth.client.id设为dummy引发KafkaException）时，异常没有进入whenComplete或exceptionally回调，而是直接抛出，如何让用户无需try/catch，仅通过whenComplete等方法处理所有错误？

Update 3：最终实现 #

RetryTemplate配置 #

@Value("${kafkaExceptions:}")
private List<String> exceptions = new ArrayList<>();

@Bean
public RetryTemplate template() throws ClassNotFoundException {
   exceptions.add(TopicAuthorizationException.class.getName());
   RetryTemplateBuilder retryTemplateBuilder = RetryTemplate.builder().maxAttempts(retries).retryOn(TopicAuthorizationException.class);
   for (String exceptionClassName : exceptions){
      retryTemplateBuilder.retryOn((Class<? extends Throwable>) Class.forName(exceptionClassName));
   }
   return retryTemplateBuilder.exponentialBackoff(1000, 2, 2000).build();
}

异步发送方法 #

@Async("exec")
public CompletableFuture<SendResult<K, V>> asyncSend(K key, String topic, V message, S) throws Exception {

   return template.execute(ctx -> {

      try {
         ProducerRecord<K, V> producerRecord = buildHeader(key, topic, msg); //可能抛出自定义异常

         return kafkaTemplate.send(producerRecord).completable();
      } catch (Exception e){
         
         if (reThrow(e)) throw e;

         return CompletableFuture.failedFuture(new CustomException(Enum.KAFKA_EX_PROD_PUBLISH_ERROR.findErrorMessage(),e));

      }
   });

}


private boolean reThrow(Exception e){
   String retriableException = e.getClass().getName();
   return retriableExceptions.contains(retriableException);
}

1. 是否可以确定：whenComplete中sr（即SendResult）非空等价于ListenableFutureCallback的onSuccess被调用？
2. 是否可以确定：whenComplete中th（即异常）非空，要么是ListenableFutureCallback的onFailure触发，要么是主动调用CompletableFuture.failedFuture()导致？

针对最初问题 #

会，完全可能返回null。因为kafkaTemplate.send()是异步操作，ListenableFuture的回调onSuccess/onFailure会在后续线程中执行，而当前线程会直接执行到return CompletableFuture.completedFuture(responseHolder[0]);，此时responseHolder[0]还未被赋值，即为null。所以客户端的whenComplete中确实可能出现result和throwable同时为null的情况。

针对Update 1 #

可以在工具库内部将ListenableFuture转换为CompletableFuture时，先完成日志记录，再将结果或异常传递给客户端：

// 示例转换逻辑
ListenableFuture<SendResult<Integer, String>> listenableFuture = kafkaTemplate.send(producerRecord);
CompletableFuture<SendResult<Integer, String>> completableFuture = new CompletableFuture<>();

listenableFuture.addCallback(
    result -> {
        log.info("success...");
        completableFuture.complete(result);
    },
    ex -> {
        log.error("error...");
        throwError(producerRecord, eventName, ex);
        completableFuture.completeExceptionally(ex);
    }
);

之后可以再对这个completableFuture进行包装，转换为客户端需要的DtoResponse类型的CompletableFuture，这样既在内部完成了日志，又能让客户端拿到明确的状态。

针对Update 2 #

直接抛出的异常是因为在template.execute()之外或者try块之外发生的同步异常（比如buildHeader抛异常但未被捕获，或者kafkaTemplate初始化异常）。要让所有异常都进入CompletableFuture的回调，需要确保所有可能抛出异常的代码都被包裹在try-catch中，并将异常转换为failedFuture：

• 把buildHeader的调用放在try块内
• 确保template.execute()本身的异常也被捕获（比如RetryTemplate达到重试次数后抛出的异常）
• 所有捕获到的异常，除了需要RetryTemplate重试的（通过reThrow判断），都转为failedFuture，而非直接抛出

针对最终确认问题 #

1. 是的。因为kafkaTemplate.send().completable()会把ListenableFuture的成功结果同步到CompletableFuture中：当ListenableFutureCallback.onSuccess被调用时，CompletableFuture会被complete(result)，此时whenComplete中的sr就是这个非空的result；反过来，如果whenComplete中的sr非空，说明CompletableFuture被正常完成，对应的就是onSuccess被触发。

2. 是的。whenComplete中的th非空的场景只有两种：

   • Kafka发送失败，触发ListenableFutureCallback.onFailure，CompletableFuture被completeExceptionally(ex)
   • 代码中主动调用CompletableFuture.failedFuture(...)，比如catch块中处理非重试异常的情况

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