private CompletableFuture<Void> sendAsync(String key, byte[] payload, Integer partition) {
            long sendTimeNanos = StatsCollector.currentNanos();
            List<Header> headers = List.of(
                new RecordHeader(HEADER_KEY_SEND_TIME_NANOS, Longs.toByteArray(sendTimeNanos))
            );
            ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic.name, partition, key, payload, headers);
            int size = payload.length;
            CompletableFuture<Void> future = new CompletableFuture<>();
            try {
                producer.send(record, (metadata, exception) -> {
                    callback.messageSent(size, sendTimeNanos, exception);
                    if (exception != null) {
                        future.completeExceptionally(exception);
                    } else {
                        future.complete(null);
                    }
                });
            } catch (InterruptException e) {
                // ignore, as we are closing
                future.completeExceptionally(e);
            } catch (Exception e) {
                callback.messageSent(size, sendTimeNanos, e);
                future.completeExceptionally(e);
            }
            return future;
        }
Send a message to the topic.

@param key       the key of the message, optional
@param payload   the payload of the message
@param partition the partition to send the message to, optional
@return a future that completes when the message is sent
tools/src/main/java/org/apache/kafka/tools/automq/perf/ProducerService.java