KeepAlive task executed for every Connection.write call

  // try to keep the connection alive if the channel allows such things - websockets will
        if (channelizer.supportsKeepAlive() && keepAliveInterval > 0) {

            final ScheduledFuture oldKeepAliveFuture = keepAliveFuture.getAndSet(cluster.executor().scheduleAtFixedRate(() -> {
                logger.debug("Request sent to server to keep {} alive", thisConnection);
                try {
                    channel.writeAndFlush(channelizer.createKeepAliveMessage());
                } catch (Exception ex) {
                    // will just log this for now - a future real request can be responsible for the failure that
                    // marks the host as dead. this also may not mean the host is actually dead. more robust handling
                    // is in play for real requests, not this simple ping
                    logger.warn(String.format("Keep-alive did not succeed on %s", thisConnection), ex);
                }
            }, keepAliveInterval, keepAliveInterval, TimeUnit.MILLISECONDS));

            // try to cancel the old future if it's still un-executed - no need to ping since a new write has come
            // through on the connection
            if (oldKeepAliveFuture != null) oldKeepAliveFuture.cancel(true);
        }

final class Connection {
    ...
    private final AtomicBoolean keepAliveInitialized = new AtomicBoolean(false);

    public ChannelPromise write(final RequestMessage requestMessage, final CompletableFuture<ResultSet> future) {
        ...
        
        // FIX HERE: with keepAliveInitialized.compareAndSet(false, true) we ensure the keepAlive task is run only once per connection.
        if (channelizer.supportsKeepAlive() && keepAliveInterval > 0 && keepAliveInitialized.compareAndSet(false, true)) {
            ...
        }
    }
}