| /* |
| * Licensed to the Apache Software Foundation (ASF) under one or more |
| * contributor license agreements. See the NOTICE file distributed with |
| * this work for additional information regarding copyright ownership. |
| * The ASF licenses this file to You under the Apache License, Version 2.0 |
| * (the "License"); you may not use this file except in compliance with |
| * the License. You may obtain a copy of the License at |
| * |
| * http://www.apache.org/licenses/LICENSE-2.0 |
| * |
| * Unless required by applicable law or agreed to in writing, software |
| * distributed under the License is distributed on an "AS IS" BASIS, |
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. |
| * See the License for the specific language governing permissions and |
| * limitations under the License. |
| */ |
| |
| package org.apache.ignite.internal; |
| |
| import java.util.Collection; |
| import java.util.UUID; |
| import java.util.concurrent.ConcurrentHashMap; |
| import java.util.concurrent.ConcurrentMap; |
| import java.util.concurrent.CountDownLatch; |
| import java.util.concurrent.atomic.AtomicInteger; |
| import java.util.concurrent.atomic.AtomicReference; |
| import org.apache.ignite.Ignite; |
| import org.apache.ignite.IgniteCheckedException; |
| import org.apache.ignite.cluster.ClusterNode; |
| import org.apache.ignite.configuration.IgniteConfiguration; |
| import org.apache.ignite.events.DiscoveryEvent; |
| import org.apache.ignite.events.Event; |
| import org.apache.ignite.events.EventType; |
| import org.apache.ignite.internal.util.typedef.F; |
| import org.apache.ignite.internal.util.typedef.X; |
| import org.apache.ignite.internal.util.typedef.internal.U; |
| import org.apache.ignite.lang.IgniteClosure; |
| import org.apache.ignite.lang.IgnitePredicate; |
| import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest; |
| import org.junit.Test; |
| |
| import static java.util.concurrent.TimeUnit.MILLISECONDS; |
| import static org.apache.ignite.events.EventType.EVT_NODE_FAILED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_JOINED; |
| import static org.apache.ignite.events.EventType.EVT_NODE_LEFT; |
| |
| /** |
| * Tests discovery event topology snapshots. |
| */ |
| public class GridDiscoveryEventSelfTest extends GridCommonAbstractTest { |
| /** Daemon flag. */ |
| private boolean daemon; |
| |
| /** {@inheritDoc} */ |
| @Override protected void beforeTest() throws Exception { |
| super.beforeTest(); |
| |
| daemon = false; |
| } |
| |
| /** */ |
| private static final IgniteClosure<ClusterNode, UUID> NODE_2ID = new IgniteClosure<ClusterNode, UUID>() { |
| @Override public UUID apply(ClusterNode n) { |
| return n.id(); |
| } |
| |
| @Override public String toString() { |
| return "Grid node shadow to node ID transformer closure."; |
| } |
| }; |
| |
| /** {@inheritDoc} */ |
| @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception { |
| IgniteConfiguration c = super.getConfiguration(igniteInstanceName); |
| |
| c.setDaemon(daemon); |
| c.setConnectorConfiguration(null); |
| |
| return c; |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testJoinSequenceEvents() throws Exception { |
| try { |
| Ignite g0 = startGrid(0); |
| |
| UUID id0 = g0.cluster().localNode().id(); |
| |
| final ConcurrentMap<Integer, Collection<ClusterNode>> evts = new ConcurrentHashMap<>(); |
| |
| final CountDownLatch latch = new CountDownLatch(3); |
| |
| g0.events().localListen(new IgnitePredicate<Event>() { |
| private AtomicInteger cnt = new AtomicInteger(); |
| |
| @Override public boolean apply(Event evt) { |
| assert evt.type() == EVT_NODE_JOINED : evt; |
| |
| evts.put(cnt.getAndIncrement(), ((DiscoveryEvent)evt).topologyNodes()); |
| |
| latch.countDown(); |
| |
| return true; |
| } |
| }, EVT_NODE_JOINED); |
| |
| UUID id1 = startGrid(1).cluster().localNode().id(); |
| UUID id2 = startGrid(2).cluster().localNode().id(); |
| UUID id3 = startGrid(3).cluster().localNode().id(); |
| |
| assertTrue("Wrong count of events received: " + evts, latch.await(3000, MILLISECONDS)); |
| |
| Collection<ClusterNode> top0 = evts.get(0); |
| |
| assertNotNull(top0); |
| assertEquals(2, top0.size()); |
| assertTrue(F.viewReadOnly(top0, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top0, NODE_2ID).contains(id1)); |
| |
| Collection<ClusterNode> top1 = evts.get(1); |
| |
| assertNotNull(top1); |
| assertEquals(3, top1.size()); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id1)); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id2)); |
| |
| Collection<ClusterNode> top2 = evts.get(2); |
| |
| assertNotNull(top2); |
| assertEquals(4, top2.size()); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id1)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id2)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id3)); |
| } |
| finally { |
| stopAllGrids(); |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testLeaveSequenceEvents() throws Exception { |
| try { |
| Ignite g0 = startGrid(0); |
| |
| UUID id0 = g0.cluster().localNode().id(); |
| UUID id1 = startGrid(1).cluster().localNode().id(); |
| UUID id2 = startGrid(2).cluster().localNode().id(); |
| UUID id3 = startGrid(3).cluster().localNode().id(); |
| |
| final ConcurrentMap<Integer, Collection<ClusterNode>> evts = new ConcurrentHashMap<>(); |
| |
| final CountDownLatch latch = new CountDownLatch(3); |
| |
| g0.events().localListen(new IgnitePredicate<Event>() { |
| private AtomicInteger cnt = new AtomicInteger(); |
| |
| @Override public boolean apply(Event evt) { |
| assert evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED : evt; |
| |
| evts.put(cnt.getAndIncrement(), ((DiscoveryEvent)evt).topologyNodes()); |
| |
| latch.countDown(); |
| |
| return true; |
| } |
| }, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| |
| stopGrid(3); |
| stopGrid(2); |
| stopGrid(1); |
| |
| assertTrue("Wrong count of events received: " + evts, latch.await(3000, MILLISECONDS)); |
| |
| Collection<ClusterNode> top2 = evts.get(0); |
| |
| assertNotNull(top2); |
| assertEquals(3, top2.size()); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id1)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top2, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top1 = evts.get(1); |
| |
| assertNotNull(top1); |
| assertEquals(2, top1.size()); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id1)); |
| assertFalse(F.viewReadOnly(top1, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top1, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top0 = evts.get(2); |
| |
| assertNotNull(top0); |
| assertEquals(1, top0.size()); |
| assertTrue(F.viewReadOnly(top0, NODE_2ID).contains(id0)); |
| assertFalse(F.viewReadOnly(top0, NODE_2ID).contains(id1)); |
| assertFalse(F.viewReadOnly(top0, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top0, NODE_2ID).contains(id3)); |
| } |
| finally { |
| stopAllGrids(); |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testMixedSequenceEvents() throws Exception { |
| try { |
| Ignite g0 = startGrid(0); |
| |
| UUID id0 = g0.cluster().localNode().id(); |
| |
| final ConcurrentMap<Integer, Collection<ClusterNode>> evts = new ConcurrentHashMap<>(); |
| |
| final CountDownLatch latch = new CountDownLatch(8); |
| |
| g0.events().localListen(new IgnitePredicate<Event>() { |
| private AtomicInteger cnt = new AtomicInteger(); |
| |
| @Override public boolean apply(Event evt) { |
| assert evt.type() == EVT_NODE_JOINED |
| || evt.type() == EVT_NODE_LEFT || evt.type() == EVT_NODE_FAILED : evt; |
| |
| evts.put(cnt.getAndIncrement(), ((DiscoveryEvent)evt).topologyNodes()); |
| |
| latch.countDown(); |
| |
| return true; |
| } |
| }, EVT_NODE_JOINED, EVT_NODE_LEFT, EVT_NODE_FAILED); |
| |
| UUID id1 = startGrid(1).cluster().localNode().id(); |
| UUID id2 = startGrid(2).cluster().localNode().id(); |
| UUID id3 = startGrid(3).cluster().localNode().id(); |
| |
| stopGrid(3); |
| stopGrid(2); |
| stopGrid(1); |
| |
| UUID id4 = startGrid(4).cluster().localNode().id(); |
| |
| stopGrid(4); |
| |
| assertTrue("Wrong count of events received [cnt= " + evts.size() + ", evts=" + evts + ']', |
| latch.await(3000, MILLISECONDS)); |
| |
| Collection<ClusterNode> top0 = evts.get(0); |
| |
| assertNotNull(top0); |
| assertEquals(2, top0.size()); |
| assertTrue(F.viewReadOnly(top0, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top0, NODE_2ID).contains(id1)); |
| |
| Collection<ClusterNode> top1 = evts.get(1); |
| |
| assertNotNull(top1); |
| assertEquals(3, top1.size()); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id1)); |
| assertTrue(F.viewReadOnly(top1, NODE_2ID).contains(id2)); |
| |
| Collection<ClusterNode> top2 = evts.get(2); |
| |
| assertNotNull(top2); |
| assertEquals(4, top2.size()); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id1)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id2)); |
| assertTrue(F.viewReadOnly(top2, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top3 = evts.get(3); |
| |
| assertNotNull(top3); |
| assertEquals(3, top3.size()); |
| assertTrue(F.viewReadOnly(top3, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top3, NODE_2ID).contains(id1)); |
| assertTrue(F.viewReadOnly(top3, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top3, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top4 = evts.get(4); |
| |
| assertNotNull(top4); |
| assertEquals(2, top4.size()); |
| assertTrue(F.viewReadOnly(top4, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top4, NODE_2ID).contains(id1)); |
| assertFalse(F.viewReadOnly(top4, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top4, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top5 = evts.get(5); |
| |
| assertNotNull(top5); |
| assertEquals(1, top5.size()); |
| assertTrue(F.viewReadOnly(top5, NODE_2ID).contains(id0)); |
| assertFalse(F.viewReadOnly(top5, NODE_2ID).contains(id1)); |
| assertFalse(F.viewReadOnly(top5, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top5, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top6 = evts.get(6); |
| |
| assertNotNull(top6); |
| assertEquals(2, top6.size()); |
| assertTrue(F.viewReadOnly(top6, NODE_2ID).contains(id0)); |
| assertTrue(F.viewReadOnly(top6, NODE_2ID).contains(id4)); |
| assertFalse(F.viewReadOnly(top6, NODE_2ID).contains(id1)); |
| assertFalse(F.viewReadOnly(top6, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top6, NODE_2ID).contains(id3)); |
| |
| Collection<ClusterNode> top7 = evts.get(7); |
| |
| assertNotNull(top7); |
| assertEquals(1, top7.size()); |
| assertTrue(F.viewReadOnly(top7, NODE_2ID).contains(id0)); |
| assertFalse(F.viewReadOnly(top7, NODE_2ID).contains(id1)); |
| assertFalse(F.viewReadOnly(top7, NODE_2ID).contains(id2)); |
| assertFalse(F.viewReadOnly(top7, NODE_2ID).contains(id3)); |
| assertFalse(F.viewReadOnly(top7, NODE_2ID).contains(id4)); |
| } |
| finally { |
| stopAllGrids(); |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testConcurrentJoinEvents() throws Exception { |
| try { |
| Ignite g0 = startGrid(0); |
| |
| UUID id0 = g0.cluster().localNode().id(); |
| |
| final ConcurrentMap<Integer, Collection<ClusterNode>> evts = new ConcurrentHashMap<>(); |
| |
| g0.events().localListen(new IgnitePredicate<Event>() { |
| private AtomicInteger cnt = new AtomicInteger(); |
| |
| @Override public boolean apply(Event evt) { |
| assert evt.type() == EVT_NODE_JOINED : evt; |
| |
| X.println(">>>>>>> Joined " + F.viewReadOnly(((DiscoveryEvent)evt).topologyNodes(), |
| NODE_2ID)); |
| |
| evts.put(cnt.getAndIncrement(), ((DiscoveryEvent)evt).topologyNodes()); |
| |
| return true; |
| } |
| }, EVT_NODE_JOINED); |
| |
| U.sleep(100); |
| |
| startGridsMultiThreaded(1, 10); |
| |
| U.sleep(100); |
| |
| assertEquals(10, evts.size()); |
| |
| for (int i = 0; i < 10; i++) { |
| Collection<ClusterNode> snapshot = evts.get(i); |
| |
| assertEquals(2 + i, snapshot.size()); |
| assertTrue(F.viewReadOnly(snapshot, NODE_2ID).contains(id0)); |
| |
| for (ClusterNode n : snapshot) |
| assertTrue("Wrong node order in snapshot [i=" + i + ", node=" + n + ']', n.order() <= 2 + i); |
| } |
| |
| Collection<UUID> ids = F.viewReadOnly(evts.get(9), NODE_2ID); |
| |
| for (int i = 1; i <= 10; i++) |
| assertTrue(ids.contains(grid(i).localNode().id())); |
| } |
| finally { |
| stopAllGrids(); |
| } |
| } |
| |
| /** |
| * @throws Exception If failed. |
| */ |
| @Test |
| public void testDaemonNodeJoin() throws Exception { |
| try { |
| startGridsMultiThreaded(3); |
| |
| final AtomicReference<IgniteCheckedException> err = new AtomicReference<>(); |
| |
| for (int i = 0; i < 3; i++) { |
| Ignite g = grid(i); |
| |
| g.events().localListen(new IgnitePredicate<Event>() { |
| @Override public boolean apply(Event evt) { |
| DiscoveryEvent discoEvt = (DiscoveryEvent)evt; |
| |
| if (discoEvt.topologyNodes().size() != 3) |
| err.compareAndSet(null, new IgniteCheckedException("Invalid discovery event [evt=" + discoEvt + |
| ", nodes=" + discoEvt.topologyNodes() + ']')); |
| |
| return true; |
| } |
| }, EventType.EVT_NODE_JOINED); |
| } |
| |
| daemon = true; |
| |
| IgniteKernal daemon = (IgniteKernal)startGrid(3); |
| |
| DiscoveryEvent join = daemon.context().discovery().localJoinEvent(); |
| |
| assertEquals(3, join.topologyNodes().size()); |
| |
| U.sleep(100); |
| |
| if (err.get() != null) |
| throw err.get(); |
| } |
| finally { |
| stopAllGrids(); |
| } |
| } |
| } |
