### Description

The `sendRequest` method is asynchronous and notifies a 
`TransportResponseHandler` when it completes the excecution
(successfully or unsuccessfully). 

When opening the connection we only catch `NodeNotConnectedException`
( https://github.com/elastic/elasticsearch/blob/main/server/src/main/java/org/elasticsearch/transport/TransportService.java#L719 ) which is currently fine as no implementations
throw other type of exception however, it'd be best to catch `Exception` here
so the `handler` gets notified of potential exceptions (as opposed to the caller
having to `try/catch` the call to `sendRequest`)
server/src/main/java/org/elasticsearch/cluster/coordination/PublicationTransportHandler.java
import org.elasticsearch.common.io.stream.RecyclerBytesStreamOutput;
import org.elasticsearch.common.io.stream.StreamInput;
import org.elasticsearch.common.io.stream.StreamOutput;
import org.elasticsearch.common.util.LazyInitializable;
import org.elasticsearch.core.AbstractRefCounted;
import org.elasticsearch.core.IOUtils;
import org.elasticsearch.core.Releasables;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.BytesTransportRequest;
import org.elasticsearch.transport.TransportException;
import org.elasticsearch.transport.TransportRequestOptions;
import org.elasticsearch.transport.TransportService;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Function;
import static org.elasticsearch.core.Strings.format;
/**
 * Implements the low-level mechanics of sending a cluster state to other nodes in the cluster during a publication.
 * <p>
 * Cluster states can be quite large and expensive to serialize, but we (mostly) send the same serialized representation to every node in
 * the cluster. This class does the serialization work once, up-front, as part of {@link #newPublicationContext} and then just re-uses the
 * resulting bytes across transport messages.
 * <p>
 * It also uses the {@link Diff} mechanism to reduce the data to be transferred wherever possible. This is only a best-effort mechanism so
 * we fall back to sending a full cluster state if the diff cannot be applied for some reason.