|
36 | 36 | import org.elasticsearch.common.settings.Setting;
|
37 | 37 | import org.elasticsearch.common.settings.Settings;
|
38 | 38 | import org.elasticsearch.common.unit.TimeValue;
|
39 |
| -import org.elasticsearch.common.util.concurrent.ConcurrentCollections; |
40 | 39 | import org.elasticsearch.discovery.zen.MembershipAction;
|
41 | 40 | import org.elasticsearch.discovery.zen.ZenDiscovery;
|
42 | 41 | import org.elasticsearch.threadpool.ThreadPool;
|
|
53 | 52 |
|
54 | 53 | import java.io.IOException;
|
55 | 54 | import java.util.Collection;
|
| 55 | +import java.util.Collections; |
56 | 56 | import java.util.HashMap;
|
| 57 | +import java.util.HashSet; |
57 | 58 | import java.util.LinkedHashMap;
|
58 | 59 | import java.util.List;
|
59 | 60 | import java.util.Map;
|
@@ -82,7 +83,7 @@ public class JoinHelper {
|
82 | 83 | private final JoinTaskExecutor joinTaskExecutor;
|
83 | 84 | private final TimeValue joinTimeout;
|
84 | 85 |
|
85 |
| - final Set<Tuple<DiscoveryNode, JoinRequest>> pendingOutgoingJoins = ConcurrentCollections.newConcurrentSet(); |
| 86 | + private final Set<Tuple<DiscoveryNode, JoinRequest>> pendingOutgoingJoins = Collections.synchronizedSet(new HashSet<>()); |
86 | 87 |
|
87 | 88 | public JoinHelper(Settings settings, AllocationService allocationService, MasterService masterService,
|
88 | 89 | TransportService transportService, LongSupplier currentTermSupplier, Supplier<ClusterState> currentStateSupplier,
|
@@ -190,8 +191,7 @@ public String toString() {
|
190 | 191 | }
|
191 | 192 |
|
192 | 193 | boolean isJoinPending() {
|
193 |
| - // cannot use pendingOutgoingJoins.isEmpty() because it's not properly synchronized. |
194 |
| - return pendingOutgoingJoins.iterator().hasNext(); |
| 194 | + return pendingOutgoingJoins.isEmpty() == false; |
195 | 195 | }
|
196 | 196 |
|
197 | 197 | public void sendJoinRequest(DiscoveryNode destination, Optional<Join> optionalJoin) {
|
|
0 commit comments