ExchangeLatchManager { public Latch getOrCreate(String id, AffinityTopologyVersion topVer) { lock.lock(); try { final CompletableLatchUid latchUid = new CompletableLatchUid(id, topVer); CompletableLatch latch = clientLatches.containsKey(latchUid) ? clientLatches.get(latchUid) : serverLatches.get(latchUid); if (latch != null) return latch; ClusterNode coordinator = getLatchCoordinator(topVer); if (coordinator == null) return null; Collection<ClusterNode> participants = getLatchParticipants(topVer); return coordinator.isLocal() ? createServerLatch(latchUid, participants) : createClientLatch(latchUid, coordinator, participants); } finally { lock.unlock(); } } ExchangeLatchManager(GridKernalContext ctx); Latch getOrCreate(String id, AffinityTopologyVersion topVer); void dropLatch(String id, AffinityTopologyVersion topVer); boolean canSkipJoiningNodes(AffinityTopologyVersion topVer); @Override String toString(); static final IgniteProductVersion PROTOCOL_V2_VERSION_SINCE; }
@Test public void shouldCorrectlyExecuteLatchWhenCrdCreatedLast() throws Exception { IgniteEx crd = startGrid(0); IgniteEx ignite1 = startGrid(1); startGrid(2); AffinityTopologyVersion nextVer = new AffinityTopologyVersion(crd.cluster().topologyVersion() + 1, 0); ignite1.context().io().sendToGridTopic( crd.localNode(), GridTopic.TOPIC_EXCHANGE, new LatchAckMessage( LATCH_NAME, nextVer, false ), GridIoPolicy.SYSTEM_POOL ); stopGrid(2); Latch latchCrdOther = latchManager(0).getOrCreate(LATCH_NAME, nextVer); latchCrdOther.countDown(); latchCrdOther.await(1, TimeUnit.SECONDS); }