Refactor a ListenableFuture (with onSuccess/onFailure) into a CompletableFuture

Because of an upgrade in the Kafka library, I need to rewrite the following code with ListenableFuture into one that uses CompletableFuture. The difference is briefly explained in this thread.

The object being handled is called SendResult<String,Object>.

The legacy code is

        ListenableFuture<SendResult<String, Object>> future = ...;
		future.addCallback(new ListenableFutureCallback<SendResult<String, Object>>() {

			@Override
			public void onSuccess(final SendResult<String, Object> result) {

				ProducerRecord<String, Object> record = result.getProducerRecord();
				CaseStatusRequest data = (CaseStatusRequest) record.value();

				logger.info("Producing request succeeded: {}", data);
			}

			@Override
			public void onFailure(final Throwable throwable) {
				logger.error("Producing request failed: {}", request.getReceiptNumber());
			}
		});

My understanding is, the analogue of ListenableFuture.onSuccess is CompletableFuture.whenComplete. That refactoring was straightforward. But the ListenableFuture.onFailure doesn't have a clear equivalent. There is CompletableFuture.exceptionally which must return something, which shouldn't be done; the error handling should be a void operation.

		CompletableFuture<SendResult<String, Object>> future = ...;
		future.whenComplete(new BiConsumer<SendResult<String,Object>,Throwable>() {

			@Override
			public void accept(SendResult<String, Object> result, Throwable u) {
				ProducerRecord<String, Object> record = result.getProducerRecord();
				CaseStatusRequest data = (CaseStatusRequest) record.value();

				logger.info("Producing request succeeded: {}", data);	
			}	
		});
		future.exceptionally(new Function<Throwable, SendResult<String,Object>>() {

			@Override
			public SendResult<String, Object> apply(Throwable arg0) {
				logger.error("Producing request failed: {}", request.getReceiptNumber());
                // Something needs to be returned here.
                // Should I return NULL?
			}
			
		});

The whenComplete function will be invoked for both success and failure cases. The signature of whenComplete takes the throwable which denotes that in case of a failure, this function will be called with the exception value.

So you may not require the exceptionally method at all.

Something like

        future.whenComplete(new BiConsumer<SendResult<String,Object>,Throwable>() {

        @Override
        public void accept(SendResult<String, Object> result, Throwable u) {
            if (u != null) {
                logger.error(....)
            }
            else {
                ProducerRecord<String, Object> record = result.getProducerRecord();
                CaseStatusRequest data = (CaseStatusRequest) record.value();

                logger.info("Producing request succeeded: {}", data);   
           }   
      }
    });

Alternative implementation

You could leverage Kafka Async send method and have a callback to handle the results. This would require some more refactoring of your older code base.

Something like this

    producer.send(producerRecord, new Callback() {
            public void onCompletion(RecordMetadata recordMetadata, Exception e) {
                // executes every time a record is successfully sent or an exception is thrown
                if (e == null) {
                    // the record was successfully sent
                    log.info("Received new metadata. \n" +
                            "Topic:" + recordMetadata.topic() + "\n" +
                            "Partition: " + recordMetadata.partition() + "\n" +
                            "Offset: " + recordMetadata.offset() + "\n" +
                            "Timestamp: " + recordMetadata.timestamp());
                } else {
                    log.error("Error while producing", e);
                }
            }
        });