34
34
import org .elasticsearch .cluster .routing .RerouteService ;
35
35
import org .elasticsearch .cluster .routing .allocation .AllocationService ;
36
36
import org .elasticsearch .cluster .service .MasterService ;
37
- import org .elasticsearch .common .Nullable ;
38
37
import org .elasticsearch .common .Priority ;
39
38
import org .elasticsearch .common .collect .Tuple ;
40
39
import org .elasticsearch .common .io .stream .StreamInput ;
43
42
import org .elasticsearch .common .unit .TimeValue ;
44
43
import org .elasticsearch .discovery .zen .MembershipAction ;
45
44
import org .elasticsearch .discovery .zen .ZenDiscovery ;
46
- import org .elasticsearch .discovery .DiscoveryModule ;
47
45
import org .elasticsearch .monitor .NodeHealthService ;
48
46
import org .elasticsearch .monitor .StatusInfo ;
49
47
import org .elasticsearch .threadpool .ThreadPool ;
@@ -83,22 +81,21 @@ public class JoinHelper {
83
81
public static final String VALIDATE_JOIN_ACTION_NAME = "internal:cluster/coordination/join/validate" ;
84
82
public static final String START_JOIN_ACTION_NAME = "internal:cluster/coordination/start_join" ;
85
83
86
- // the timeout for each join attempt
84
+ // the timeout for Zen1 join attempts
87
85
public static final Setting <TimeValue > JOIN_TIMEOUT_SETTING =
88
86
Setting .timeSetting ("cluster.join.timeout" ,
89
- TimeValue .timeValueMillis (60000 ), TimeValue .timeValueMillis (1 ), Setting .Property .NodeScope );
87
+ TimeValue .timeValueMillis (60000 ), TimeValue .timeValueMillis (1 ), Setting .Property .NodeScope , Setting . Property . Deprecated );
90
88
91
89
private final MasterService masterService ;
92
90
private final TransportService transportService ;
93
91
private final JoinTaskExecutor joinTaskExecutor ;
94
92
95
- @ Nullable // if using single-node discovery
96
- private final TimeValue joinTimeout ;
93
+ private final TimeValue joinTimeout ; // only used for Zen1 joining
97
94
private final NodeHealthService nodeHealthService ;
98
95
99
96
private final Set <Tuple <DiscoveryNode , JoinRequest >> pendingOutgoingJoins = Collections .synchronizedSet (new HashSet <>());
100
97
101
- private AtomicReference <FailedJoinAttempt > lastFailedJoinAttempt = new AtomicReference <>();
98
+ private final AtomicReference <FailedJoinAttempt > lastFailedJoinAttempt = new AtomicReference <>();
102
99
103
100
JoinHelper (Settings settings , AllocationService allocationService , MasterService masterService ,
104
101
TransportService transportService , LongSupplier currentTermSupplier , Supplier <ClusterState > currentStateSupplier ,
@@ -108,7 +105,7 @@ public class JoinHelper {
108
105
this .masterService = masterService ;
109
106
this .transportService = transportService ;
110
107
this .nodeHealthService = nodeHealthService ;
111
- this .joinTimeout = DiscoveryModule . isSingleNodeDiscovery ( settings ) ? null : JOIN_TIMEOUT_SETTING .get (settings );
108
+ this .joinTimeout = JOIN_TIMEOUT_SETTING .get (settings );
112
109
this .joinTaskExecutor = new JoinTaskExecutor (settings , allocationService , logger , rerouteService ) {
113
110
114
111
@ Override
@@ -286,15 +283,17 @@ public void sendJoinRequest(DiscoveryNode destination, long term, Optional<Join>
286
283
logger .debug ("attempting to join {} with {}" , destination , joinRequest );
287
284
final String actionName ;
288
285
final TransportRequest transportRequest ;
286
+ final TransportRequestOptions transportRequestOptions ;
289
287
if (Coordinator .isZen1Node (destination )) {
290
288
actionName = MembershipAction .DISCOVERY_JOIN_ACTION_NAME ;
291
289
transportRequest = new MembershipAction .JoinRequest (transportService .getLocalNode ());
290
+ transportRequestOptions = TransportRequestOptions .builder ().withTimeout (joinTimeout ).build ();
292
291
} else {
293
292
actionName = JOIN_ACTION_NAME ;
294
293
transportRequest = joinRequest ;
294
+ transportRequestOptions = TransportRequestOptions .EMPTY ;
295
295
}
296
- transportService .sendRequest (destination , actionName , transportRequest ,
297
- TransportRequestOptions .builder ().withTimeout (joinTimeout ).build (),
296
+ transportService .sendRequest (destination , actionName , transportRequest , transportRequestOptions ,
298
297
new TransportResponseHandler <Empty >() {
299
298
@ Override
300
299
public Empty read (StreamInput in ) {
@@ -363,9 +362,7 @@ public void sendValidateJoinRequest(DiscoveryNode node, ClusterState state, Acti
363
362
} else {
364
363
actionName = VALIDATE_JOIN_ACTION_NAME ;
365
364
}
366
- transportService .sendRequest (node , actionName ,
367
- new ValidateJoinRequest (state ),
368
- TransportRequestOptions .builder ().withTimeout (joinTimeout ).build (),
365
+ transportService .sendRequest (node , actionName , new ValidateJoinRequest (state ),
369
366
new ActionListenerResponseHandler <>(listener , i -> Empty .INSTANCE , ThreadPool .Names .GENERIC ));
370
367
}
371
368
0 commit comments