Kafka message default ack npe fix

[danielkec]

As @tkote reported on Slack, when KafkaMessage is created with KafkaMessage.of(key, value), only 5 messages are sent and no more is ever requested.
Problem is caused by wrong initialization of the ack future supplier.

    static <K, V> KafkaMessage<K, V> of(K key, V payload) {
        Objects.requireNonNull(payload);
        return new KafkaProducerMessage<>(key, payload, null);
    }

So Kafka ack callback silently dies and nothing is ever requested again:

        CompletableFuture.allOf(futureList.toArray(new CompletableFuture[0]))
                .whenComplete((success, exception) -> {
                    if (exception == null) {
                        message.ack().whenComplete((a, b) -> {
                            // Atomically increment
                            // or reset backpressureCounter if incrementing would reach threshold
                            if (backpressureCounter.getAndUpdate(n -> ++n == backpressure ? 0 : n)
                                    >= backpressure - 1) {
                                // configured backpressure threshold reached
                                subscription.request(backpressure);
                            }
                        });
                    }
                });

Signed-off-by: Daniel Kec daniel.kec@oracle.com