Cross-cluster search and default connections can get crossed

[jasontedor]

This is arising during recovery where it appears that the cross-cluster search connection profile is being selected:

Caused by: java.lang.IllegalStateException: can't select channel size is 0 for types: [RECOVERY, BULK, STATE]
	at org.elasticsearch.transport.ConnectionProfile$ConnectionTypeHandle.getChannel(ConnectionProfile.java:216) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TcpTransport$NodeChannels.channel(TcpTransport.java:416) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TcpTransport$NodeChannels.sendRequest(TcpTransport.java:441) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TransportService.sendRequestInternal(TransportService.java:586) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TransportService.sendRequest(TransportService.java:519) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TransportService.submitRequest(TransportService.java:481) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RemoteRecoveryTargetHandler.writeFileChunk(RemoteRecoveryTargetHandler.java:151) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RecoverySourceHandler$RecoveryOutputStream.lambda$sendNextChunk$0(RecoverySourceHandler.java:537) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.common.util.CancellableThreads.executeIO(CancellableThreads.java:105) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RecoverySourceHandler$RecoveryOutputStream.sendNextChunk(RecoverySourceHandler.java:536) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RecoverySourceHandler$RecoveryOutputStream.write(RecoverySourceHandler.java:529) ~[elasticsearch-5.6.3.jar:5.6.3]
	at java.io.BufferedOutputStream.flushBuffer(BufferedOutputStream.java:82) ~[?:1.8.0_144]
	at java.io.BufferedOutputStream.flush(BufferedOutputStream.java:140) ~[?:1.8.0_144]
	at org.elasticsearch.common.io.Streams.copy(Streams.java:83) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.common.io.Streams.copy(Streams.java:60) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RecoverySourceHandler.sendFiles(RecoverySourceHandler.java:554) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RecoverySourceHandler.phase1(RecoverySourceHandler.java:276) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.RecoverySourceHandler.recoverToTarget(RecoverySourceHandler.java:138) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.PeerRecoverySourceService.recover(PeerRecoverySourceService.java:132) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.PeerRecoverySourceService.access$100(PeerRecoverySourceService.java:54) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.PeerRecoverySourceService$StartRecoveryTransportRequestHandler.messageReceived(PeerRecoverySourceService.java:141) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.indices.recovery.PeerRecoverySourceService$StartRecoveryTransportRequestHandler.messageReceived(PeerRecoverySourceService.java:138) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TransportRequestHandler.messageReceived(TransportRequestHandler.java:33) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.RequestHandlerRegistry.processMessageReceived(RequestHandlerRegistry.java:69) ~[elasticsearch-5.6.3.jar:5.6.3]
	at org.elasticsearch.transport.TcpTransport$RequestHandler.doRun(TcpTransport.java:1539) ~[elasticsearch-5.6.3.jar:5.6.3]
	... 5 more

[elasticmachine]

Pinging @elastic/es-core-infra

[Tim-Brooks]

I looked at this today. It appears to me that the majority of connections use the TransportService openConnection method to open the connection (which is not managed by the TransportService). However, one node is picked as the "handshake node". This node is officially added to the TransportService through the connectToNode method. I think that this means that the cross-cluster search profile is stored in the TransportService and used for that node's connection (if it is registered first).

I'm not super familiar with the cross-cluster search design. But I assume registering the handshake node with the TransportService is intentional @s1monw?

We should consider:

1. If we want cross-cluster search nodes to be registered with the TransportService.
2. Adding validation to prevent adding your the current cluster as a cluster for cross-cluster search.

In RemoteClusterConnection:

     try {
        handshakeNode = transportService.handshake(connection, remoteProfile.getHandshakeTimeout().millis(),
            (c) -> remoteClusterName.get() == null ? true : c.equals(remoteClusterName.get()));
    } catch (IllegalStateException ex) {
        logger.warn((Supplier<?>) () -> new ParameterizedMessage("seed node {} cluster name mismatch expected " +
            "cluster name {}", connection.getNode(), remoteClusterName.get()), ex);
        throw ex;
    }
                            if (nodePredicate.test(handshakeNode) && connectedNodes.size() < maxNumRemoteConnections) {
        transportService.connectToNode(handshakeNode, remoteProfile);
        connectedNodes.add(handshakeNode);
    }