Skip to content

Commit 9e7fd8c

Browse files
andrershovandrershov
authored
Migrate ZenDiscoveryIT to Zen2 (#37465)
ZenDiscoveryIT contained 5 tests. 3 run without changes, testNodeRejectsClusterStateWithWrongMasterNode removed, testHandleNodeJoin_incompatibleClusterState changed.
1 parent 170d741 commit 9e7fd8c

File tree

2 files changed

+50
-98
lines changed

2 files changed

+50
-98
lines changed

‎server/src/main/java/org/elasticsearch/cluster/coordination/Coordinator.java‎

Lines changed: 26 additions & 18 deletions
Original file line numberDiff line numberDiff line change
@@ -397,6 +397,7 @@ private Join joinLeaderInTerm(StartJoinRequest startJoinRequest) {
397397
}
398398
}
399399

400+
400401
private void handleJoinRequest(JoinRequest joinRequest, JoinHelper.JoinCallback joinCallback) {
401402
assert Thread.holdsLock(mutex) == false;
402403
assert getLocalNode().isMasterNode() : getLocalNode() + " received a join but is not master-eligible";
@@ -413,30 +414,37 @@ private void handleJoinRequest(JoinRequest joinRequest, JoinHelper.JoinCallback
413414
JoinTaskExecutor.ensureMajorVersionBarrier(joinRequest.getSourceNode().getVersion(),
414415
stateForJoinValidation.getNodes().getMinNodeVersion());
415416
}
417+
sendValidateJoinRequest(stateForJoinValidation, joinRequest, joinCallback);
416418

417-
// validate the join on the joining node, will throw a failure if it fails the validation
418-
joinHelper.sendValidateJoinRequest(joinRequest.getSourceNode(), stateForJoinValidation, new ActionListener<Empty>() {
419-
@Override
420-
public void onResponse(Empty empty) {
421-
try {
422-
processJoinRequest(joinRequest, joinCallback);
423-
} catch (Exception e) {
424-
joinCallback.onFailure(e);
425-
}
426-
}
427-
428-
@Override
429-
public void onFailure(Exception e) {
430-
logger.warn(() -> new ParameterizedMessage("failed to validate incoming join request from node [{}]",
431-
joinRequest.getSourceNode()), e);
432-
joinCallback.onFailure(new IllegalStateException("failure when sending a validation request to node", e));
433-
}
434-
});
435419
} else {
436420
processJoinRequest(joinRequest, joinCallback);
437421
}
438422
}
439423

424+
// package private for tests
425+
void sendValidateJoinRequest(ClusterState stateForJoinValidation, JoinRequest joinRequest,
426+
JoinHelper.JoinCallback joinCallback) {
427+
// validate the join on the joining node, will throw a failure if it fails the validation
428+
joinHelper.sendValidateJoinRequest(joinRequest.getSourceNode(), stateForJoinValidation, new ActionListener<Empty>() {
429+
@Override
430+
public void onResponse(Empty empty) {
431+
try {
432+
processJoinRequest(joinRequest, joinCallback);
433+
} catch (Exception e) {
434+
joinCallback.onFailure(e);
435+
}
436+
}
437+
438+
@Override
439+
public void onFailure(Exception e) {
440+
logger.warn(() -> new ParameterizedMessage("failed to validate incoming join request from node [{}]",
441+
joinRequest.getSourceNode()), e);
442+
joinCallback.onFailure(new IllegalStateException("failure when sending a validation request to node", e));
443+
}
444+
});
445+
}
446+
447+
440448
private void processJoinRequest(JoinRequest joinRequest, JoinHelper.JoinCallback joinCallback) {
441449
final Optional<Join> optionalJoin = joinRequest.getOptionalJoin();
442450
synchronized (mutex) {

‎server/src/test/java/org/elasticsearch/discovery/zen/ZenDiscoveryIT.java‎ renamed to ‎server/src/test/java/org/elasticsearch/cluster/coordination/ZenDiscoveryIT.java‎

Lines changed: 24 additions & 80 deletions
Original file line numberDiff line numberDiff line change
@@ -17,51 +17,41 @@
1717
* under the License.
1818
*/
1919

20-
package org.elasticsearch.discovery.zen;
20+
package org.elasticsearch.cluster.coordination;
2121

22-
import org.elasticsearch.ExceptionsHelper;
2322
import org.elasticsearch.Version;
2423
import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
2524
import org.elasticsearch.action.admin.cluster.node.stats.NodesStatsResponse;
2625
import org.elasticsearch.action.admin.indices.recovery.RecoveryResponse;
2726
import org.elasticsearch.cluster.ClusterState;
2827
import org.elasticsearch.cluster.metadata.MetaData;
2928
import org.elasticsearch.cluster.node.DiscoveryNode;
30-
import org.elasticsearch.cluster.node.DiscoveryNodes;
3129
import org.elasticsearch.cluster.service.ClusterService;
3230
import org.elasticsearch.common.Priority;
3331
import org.elasticsearch.common.Strings;
34-
import org.elasticsearch.common.bytes.BytesReference;
3532
import org.elasticsearch.common.settings.Settings;
3633
import org.elasticsearch.common.xcontent.ToXContent;
3734
import org.elasticsearch.common.xcontent.XContentBuilder;
3835
import org.elasticsearch.common.xcontent.XContentFactory;
3936
import org.elasticsearch.discovery.Discovery;
4037
import org.elasticsearch.discovery.DiscoveryStats;
38+
import org.elasticsearch.discovery.zen.FaultDetection;
4139
import org.elasticsearch.node.Node;
4240
import org.elasticsearch.test.ESIntegTestCase;
4341
import org.elasticsearch.test.TestCustomMetaData;
44-
import org.elasticsearch.test.discovery.TestZenDiscovery;
4542
import org.elasticsearch.test.junit.annotations.TestLogging;
46-
import org.elasticsearch.threadpool.ThreadPool;
47-
import org.elasticsearch.transport.BytesTransportRequest;
48-
import org.elasticsearch.transport.EmptyTransportResponseHandler;
49-
import org.elasticsearch.transport.TransportException;
50-
import org.elasticsearch.transport.TransportResponse;
51-
import org.elasticsearch.transport.TransportService;
43+
import org.elasticsearch.transport.RemoteTransportException;
5244

5345
import java.io.IOException;
54-
import java.net.UnknownHostException;
55-
import java.util.ArrayList;
5646
import java.util.EnumSet;
57-
import java.util.List;
47+
import java.util.Optional;
48+
import java.util.concurrent.CompletableFuture;
5849
import java.util.concurrent.CountDownLatch;
5950
import java.util.concurrent.ExecutionException;
51+
import java.util.concurrent.TimeUnit;
52+
import java.util.concurrent.TimeoutException;
6053
import java.util.concurrent.atomic.AtomicInteger;
61-
import java.util.concurrent.atomic.AtomicReference;
6254

63-
import static java.util.Collections.emptyMap;
64-
import static java.util.Collections.emptySet;
6555
import static org.hamcrest.Matchers.containsString;
6656
import static org.hamcrest.Matchers.either;
6757
import static org.hamcrest.Matchers.equalTo;
@@ -73,13 +63,6 @@
7363
@TestLogging("_root:DEBUG")
7464
public class ZenDiscoveryIT extends ESIntegTestCase {
7565

76-
@Override
77-
protected Settings nodeSettings(int nodeOrdinal) {
78-
return Settings.builder().put(super.nodeSettings(nodeOrdinal))
79-
.put(TestZenDiscovery.USE_ZEN2.getKey(), false) // Zen1-specific stuff in some tests
80-
.build();
81-
}
82-
8366
public void testNoShardRelocationsOccurWhenElectedMasterNodeFails() throws Exception {
8467
Settings defaultSettings = Settings.builder()
8568
.put(FaultDetection.PING_TIMEOUT_SETTING.getKey(), "1s")
@@ -122,7 +105,7 @@ public void testNoShardRelocationsOccurWhenElectedMasterNodeFails() throws Excep
122105
assertThat(numRecoveriesAfterNewMaster, equalTo(numRecoveriesBeforeNewMaster));
123106
}
124107

125-
public void testNodeFailuresAreProcessedOnce() throws ExecutionException, InterruptedException, IOException {
108+
public void testNodeFailuresAreProcessedOnce() throws IOException {
126109
Settings defaultSettings = Settings.builder()
127110
.put(FaultDetection.PING_TIMEOUT_SETTING.getKey(), "1s")
128111
.put(FaultDetection.PING_RETRIES_SETTING.getKey(), "1")
@@ -161,78 +144,39 @@ public void testNodeFailuresAreProcessedOnce() throws ExecutionException, Interr
161144
assertThat(numUpdates.get(), either(equalTo(1)).or(equalTo(2))); // due to batching, both nodes can be handled in same CS update
162145
}
163146

164-
public void testNodeRejectsClusterStateWithWrongMasterNode() throws Exception {
165-
List<String> nodeNames = internalCluster().startNodes(2);
166-
167-
List<String> nonMasterNodes = new ArrayList<>(nodeNames);
168-
nonMasterNodes.remove(internalCluster().getMasterName());
169-
String noneMasterNode = nonMasterNodes.get(0);
170-
171-
ClusterState state = internalCluster().getInstance(ClusterService.class).state();
172-
DiscoveryNode node = null;
173-
for (DiscoveryNode discoveryNode : state.nodes()) {
174-
if (discoveryNode.getName().equals(noneMasterNode)) {
175-
node = discoveryNode;
176-
}
177-
}
178-
assert node != null;
179-
180-
DiscoveryNodes.Builder nodes = DiscoveryNodes.builder(state.nodes())
181-
.add(new DiscoveryNode("abc", buildNewFakeTransportAddress(), emptyMap(),
182-
emptySet(), Version.CURRENT)).masterNodeId("abc");
183-
ClusterState.Builder builder = ClusterState.builder(state);
184-
builder.nodes(nodes);
185-
BytesReference bytes = PublishClusterStateAction.serializeFullClusterState(builder.build(), node.getVersion());
186-
187-
final CountDownLatch latch = new CountDownLatch(1);
188-
final AtomicReference<Exception> reference = new AtomicReference<>();
189-
internalCluster().getInstance(TransportService.class, noneMasterNode).sendRequest(node, PublishClusterStateAction.SEND_ACTION_NAME,
190-
new BytesTransportRequest(bytes, Version.CURRENT), new EmptyTransportResponseHandler(ThreadPool.Names.SAME) {
191-
192-
@Override
193-
public void handleResponse(TransportResponse.Empty response) {
194-
super.handleResponse(response);
195-
latch.countDown();
196-
}
197-
198-
@Override
199-
public void handleException(TransportException exp) {
200-
super.handleException(exp);
201-
reference.set(exp);
202-
latch.countDown();
203-
}
204-
});
205-
latch.await();
206-
assertThat(reference.get(), notNullValue());
207-
assertThat(ExceptionsHelper.detailedMessage(reference.get()),
208-
containsString("cluster state from a different master than the current one, rejecting"));
209-
}
210-
211-
public void testHandleNodeJoin_incompatibleClusterState() throws UnknownHostException {
212-
String masterOnlyNode = internalCluster().startMasterOnlyNode();
147+
public void testHandleNodeJoin_incompatibleClusterState()
148+
throws InterruptedException, ExecutionException, TimeoutException {
149+
String masterNode = internalCluster().startMasterOnlyNode();
213150
String node1 = internalCluster().startNode();
214-
ZenDiscovery zenDiscovery = (ZenDiscovery) internalCluster().getInstance(Discovery.class, masterOnlyNode);
215151
ClusterService clusterService = internalCluster().getInstance(ClusterService.class, node1);
152+
Coordinator coordinator = (Coordinator) internalCluster().getInstance(Discovery.class, masterNode);
216153
final ClusterState state = clusterService.state();
217154
MetaData.Builder mdBuilder = MetaData.builder(state.metaData());
218155
mdBuilder.putCustom(CustomMetaData.TYPE, new CustomMetaData("data"));
219156
ClusterState stateWithCustomMetaData = ClusterState.builder(state).metaData(mdBuilder).build();
220157

221-
final AtomicReference<IllegalStateException> holder = new AtomicReference<>();
158+
final CompletableFuture<Throwable> future = new CompletableFuture<>();
222159
DiscoveryNode node = state.nodes().getLocalNode();
223-
zenDiscovery.handleJoinRequest(node, stateWithCustomMetaData, new MembershipAction.JoinCallback() {
160+
161+
coordinator.sendValidateJoinRequest(stateWithCustomMetaData, new JoinRequest(node, Optional.empty()),
162+
new JoinHelper.JoinCallback() {
224163
@Override
225164
public void onSuccess() {
165+
future.completeExceptionally(new AssertionError("onSuccess should not be called"));
226166
}
227167

228168
@Override
229169
public void onFailure(Exception e) {
230-
holder.set((IllegalStateException) e);
170+
future.complete(e);
231171
}
232172
});
233173

234-
assertThat(holder.get(), notNullValue());
235-
assertThat(holder.get().getMessage(), equalTo("failure when sending a validation request to node"));
174+
Throwable t = future.get(10, TimeUnit.SECONDS);
175+
176+
assertTrue(t instanceof IllegalStateException);
177+
assertTrue(t.getCause() instanceof RemoteTransportException);
178+
assertTrue(t.getCause().getCause() instanceof IllegalArgumentException);
179+
assertThat(t.getCause().getCause().getMessage(), containsString("Unknown NamedWriteable"));
236180
}
237181

238182
public static class CustomMetaData extends TestCustomMetaData {

0 commit comments

Comments
 (0)