### 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
false,
            BytesTransportRequest::new,
            (request, channel, task) -> channel.sendResponse(handleIncomingPublishRequest(request))
        );
    }
    public PublishClusterStateStats stats() {
        return new PublishClusterStateStats(
            fullClusterStateReceivedCount.get(),
            incompatibleClusterStateDiffReceivedCount.get(),
            compatibleClusterStateDiffReceivedCount.get(),
            serializationStatsTracker.getSerializationStats()
        );
    }
    private PublishWithJoinResponse handleIncomingPublishRequest(BytesTransportRequest request) throws IOException {
        final Compressor compressor = CompressorFactory.compressor(request.bytes());
        StreamInput in = request.bytes().streamInput();
        try {
            if (compressor != null) {
                in = new InputStreamStreamInput(compressor.threadLocalInputStream(in));
            }
            in = new NamedWriteableAwareStreamInput(in, namedWriteableRegistry);
            in.setVersion(request.version());
            
            if (in.readBoolean()) {
                final ClusterState incomingState;
                
                try (StreamInput input = in) {
                    incomingState = ClusterState.readFrom(input, transportService.getLocalNode());
                } catch (Exception e) {
                    logger.warn("unexpected error while deserializing an incoming cluster state", e);