# IGNITE-45 - Fixing tests
Project: http://git-wip-us.apache.org/repos/asf/incubator-ignite/repo Commit: http://git-wip-us.apache.org/repos/asf/incubator-ignite/commit/65d22d58 Tree: http://git-wip-us.apache.org/repos/asf/incubator-ignite/tree/65d22d58 Diff: http://git-wip-us.apache.org/repos/asf/incubator-ignite/diff/65d22d58 Branch: refs/heads/ignite-45 Commit: 65d22d5861eafda5912fb83df197df35024a8f9c Parents: 25e1fbd Author: Alexey Goncharuk <agoncha...@gridgain.com> Authored: Wed Mar 11 22:35:06 2015 -0700 Committer: Alexey Goncharuk <agoncha...@gridgain.com> Committed: Wed Mar 11 22:35:06 2015 -0700 ---------------------------------------------------------------------- .../configuration/IgniteConfiguration.java | 1 + .../apache/ignite/internal/IgniteKernal.java | 33 +++- .../org/apache/ignite/internal/IgnitionEx.java | 1 + .../internal/cluster/ClusterGroupAdapter.java | 6 + .../cache/GridCacheExplicitLockSpan.java | 20 +- .../processors/cache/GridCacheMvccManager.java | 16 +- .../processors/cache/GridCacheProcessor.java | 192 +++++++++++-------- .../colocated/GridDhtColocatedLockFuture.java | 67 +++---- .../distributed/near/GridNearLockFuture.java | 64 +++---- .../cache/distributed/near/GridNearTxLocal.java | 22 +-- .../near/GridNearTxPrepareFuture.java | 19 -- .../cache/transactions/IgniteInternalTx.java | 5 + .../cache/transactions/IgniteTxAdapter.java | 13 +- .../cache/version/GridCacheVersionManager.java | 13 +- .../datastructures/DataStructuresProcessor.java | 2 +- .../spi/discovery/tcp/TcpDiscoverySpi.java | 2 + .../cache/GridCacheAbstractFlagsTest.java | 7 +- .../cache/GridCacheAbstractSelfTest.java | 18 +- ...ridCacheMultinodeUpdateAbstractSelfTest.java | 16 +- .../cache/IgniteDynamicCacheStartSelfTest.java | 14 +- .../GridCacheClientModesAbstractSelfTest.java | 5 - ...idCacheNearOnlyMultiNodeFullApiSelfTest.java | 28 ++- ...dCacheAtomicLocalMetricsNoStoreSelfTest.java | 11 +- 23 files changed, 306 insertions(+), 269 deletions(-) ---------------------------------------------------------------------- http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java index 031fdec..c2bfce4 100644 --- a/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java +++ b/modules/core/src/main/java/org/apache/ignite/configuration/IgniteConfiguration.java @@ -431,6 +431,7 @@ public class IgniteConfiguration { cacheSanityCheckEnabled = cfg.isCacheSanityCheckEnabled(); connectorCfg = cfg.getConnectorConfiguration(); classLdr = cfg.getClassLoader(); + clientMode = cfg.isClientMode(); clockSyncFreq = cfg.getClockSyncFrequency(); clockSyncSamples = cfg.getClockSyncSamples(); deployMode = cfg.getDeploymentMode(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java index b4b4c8b..cac8666 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgniteKernal.java @@ -2258,7 +2258,7 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable { guard(); try { - ctx.cache().dynamicStartCache(cacheCfg).get(); + ctx.cache().dynamicStartCache(cacheCfg, null).get(); return ctx.cache().publicJCache(cacheCfg.getName()); } @@ -2273,14 +2273,37 @@ public class IgniteKernal implements IgniteEx, IgniteMXBean, Externalizable { /** {@inheritDoc} */ @Override public <K, V> IgniteCache<K, V> createCache(CacheConfiguration<K, V> cacheCfg, @Nullable NearCacheConfiguration<K, V> nearCfg) { - // TODO: implement. - return null; + guard(); + + try { + ctx.cache().dynamicStartCache(cacheCfg, nearCfg).get(); + + return ctx.cache().publicJCache(cacheCfg.getName()); + } + catch (IgniteCheckedException e) { + throw new CacheException(e); + } + finally { + unguard(); + } + } /** {@inheritDoc} */ @Override public <K, V> IgniteCache<K, V> createCache(@Nullable NearCacheConfiguration<K, V> nearCfg) { - // TODO: implement. - return null; + guard(); + + try { + ctx.cache().dynamicStartCache(null, nearCfg).get(); + + return ctx.cache().publicJCache(nearCfg.getName()); + } + catch (IgniteCheckedException e) { + throw new CacheException(e); + } + finally { + unguard(); + } } /** {@inheritDoc} */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java index 3c8e953..deb7707 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/IgnitionEx.java @@ -1915,6 +1915,7 @@ public class IgnitionEx { cache.setPreloadMode(SYNC); cache.setWriteSynchronizationMode(FULL_SYNC); cache.setAffinity(new CacheRendezvousAffinityFunction(false, 100)); + cache.setNodeFilter(CacheConfiguration.ALL_NODES); return cache; } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java index e955d81..529bee2 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/cluster/ClusterGroupAdapter.java @@ -333,8 +333,14 @@ public class ClusterGroupAdapter implements ClusterGroupEx, Externalizable { guard(); try { + if (p != null) + ctx.resource().injectGeneric(p); + return new ClusterGroupAdapter(ctx, subjId, this.p != null ? F.and(p, this.p) : p); } + catch (IgniteCheckedException e) { + throw U.convertException(e); + } finally { unguard(); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheExplicitLockSpan.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheExplicitLockSpan.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheExplicitLockSpan.java index a468fb9..242e769 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheExplicitLockSpan.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheExplicitLockSpan.java @@ -18,7 +18,7 @@ package org.apache.ignite.internal.processors.cache; import org.apache.ignite.internal.*; -import org.apache.ignite.internal.managers.discovery.*; +import org.apache.ignite.internal.processors.affinity.*; import org.apache.ignite.internal.processors.cache.version.*; import org.apache.ignite.internal.util.future.*; import org.apache.ignite.internal.util.tostring.*; @@ -38,7 +38,7 @@ public class GridCacheExplicitLockSpan extends ReentrantLock { /** Topology snapshot. */ @GridToStringInclude - private final GridDiscoveryTopologySnapshot topSnapshot; + private final AffinityTopologyVersion topVer; /** Pending candidates. */ @GridToStringInclude @@ -49,11 +49,11 @@ public class GridCacheExplicitLockSpan extends ReentrantLock { private final GridFutureAdapter<Object> releaseFut = new GridFutureAdapter<>(); /** - * @param topSnapshot Topology snapshot. + * @param topVer Topology version. * @param cand Candidate. */ - public GridCacheExplicitLockSpan(GridDiscoveryTopologySnapshot topSnapshot, GridCacheMvccCandidate cand) { - this.topSnapshot = topSnapshot; + public GridCacheExplicitLockSpan(AffinityTopologyVersion topVer, GridCacheMvccCandidate cand) { + this.topVer = topVer; ensureDeque(cand.key()).addFirst(cand); } @@ -61,19 +61,19 @@ public class GridCacheExplicitLockSpan extends ReentrantLock { /** * Adds candidate to a lock span. * - * @param topSnapshot Topology snapshot for which candidate is added. + * @param topVer Topology snapshot for which candidate is added. * @param cand Candidate to add. * @return {@code True} if candidate was added, {@code false} if this span is empty and * new span should be created. */ - public boolean addCandidate(GridDiscoveryTopologySnapshot topSnapshot, GridCacheMvccCandidate cand) { + public boolean addCandidate(AffinityTopologyVersion topVer, GridCacheMvccCandidate cand) { lock(); try { if (cands.isEmpty()) return false; - assert this.topSnapshot.topologyVersion() == topSnapshot.topologyVersion(); + assert this.topVer.equals(this.topVer); Deque<GridCacheMvccCandidate> deque = ensureDeque(cand.key()); @@ -234,8 +234,8 @@ public class GridCacheExplicitLockSpan extends ReentrantLock { * * @return Topology snapshot or {@code null} if candidate list is empty. */ - @Nullable public GridDiscoveryTopologySnapshot topologySnapshot() { - return releaseFut.isDone() ? null : topSnapshot; + @Nullable public AffinityTopologyVersion topologyVersion() { + return releaseFut.isDone() ? null : topVer; } /** http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java index 5725d6c..2ca8986 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheMvccManager.java @@ -736,14 +736,14 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter { * * @param threadId Thread ID. * @param cand Candidate to add. - * @param snapshot Topology snapshot. + * @param topVer Topology version. */ - public void addExplicitLock(long threadId, GridCacheMvccCandidate cand, GridDiscoveryTopologySnapshot snapshot) { + public void addExplicitLock(long threadId, GridCacheMvccCandidate cand, AffinityTopologyVersion topVer) { while (true) { GridCacheExplicitLockSpan span = pendingExplicit.get(cand.threadId()); if (span == null) { - span = new GridCacheExplicitLockSpan(snapshot, cand); + span = new GridCacheExplicitLockSpan(topVer, cand); GridCacheExplicitLockSpan old = pendingExplicit.putIfAbsent(threadId, span); @@ -754,7 +754,7 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter { } // Either span was not empty, or concurrent put did not succeed. - if (span.addCandidate(snapshot, cand)) + if (span.addCandidate(topVer, cand)) break; else pendingExplicit.remove(threadId, span); @@ -887,10 +887,10 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter { * @param threadId Thread ID. * @return Topology snapshot for last acquired and not released lock. */ - @Nullable public GridDiscoveryTopologySnapshot lastExplicitLockTopologySnapshot(long threadId) { + @Nullable public AffinityTopologyVersion lastExplicitLockTopologyVersion(long threadId) { GridCacheExplicitLockSpan span = pendingExplicit.get(threadId); - return span != null ? span.topologySnapshot() : null; + return span != null ? span.topologyVersion() : null; } /** {@inheritDoc} */ @@ -944,9 +944,9 @@ public class GridCacheMvccManager extends GridCacheSharedManagerAdapter { GridCompoundFuture<Object, Object> res = new GridCompoundFuture<>(); for (GridCacheExplicitLockSpan span : pendingExplicit.values()) { - GridDiscoveryTopologySnapshot snapshot = span.topologySnapshot(); + AffinityTopologyVersion snapshot = span.topologyVersion(); - if (snapshot != null && snapshot.topologyVersion() < topVer.topologyVersion()) + if (snapshot != null && snapshot.compareTo(topVer) < 0) res.add(span.releaseFuture()); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java index 83397b0..758a4f8 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/GridCacheProcessor.java @@ -75,6 +75,9 @@ import static org.apache.ignite.transactions.TransactionIsolation.*; */ @SuppressWarnings("unchecked") public class GridCacheProcessor extends GridProcessorAdapter { + /** Null cache name. */ + private static final String NULL_NAME = U.id8(UUID.randomUUID()); + /** Shared cache context. */ private GridCacheSharedContext<?, ?> sharedCtx; @@ -100,7 +103,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { private final Set<String> sysCaches; /** Caches stop sequence. */ - private final Deque<GridCacheAdapter<?, ?>> stopSeq; + private final Deque<String> stopSeq; /** Transaction interface implementation. */ private IgniteTransactionsImpl transactions; @@ -590,8 +593,6 @@ public class GridCacheProcessor extends GridProcessorAdapter { ctx.performance().add("Disable serializable transactions (set 'txSerializableEnabled' to false)", !ctx.config().getTransactionConfiguration().isTxSerializableEnabled()); - Collection<GridCacheAdapter<?, ?>> startSeq = new ArrayList<>(cfgs.length); - for (int i = 0; i < cfgs.length; i++) { CacheConfiguration<?, ?> cfg = new CacheConfiguration(cfgs[i]); @@ -613,13 +614,11 @@ public class GridCacheProcessor extends GridProcessorAdapter { "assign unique name to each cache)."); } - GridCacheContext cacheCtx = createCache(cfg, cacheObjCtx); - DynamicCacheDescriptor desc = new DynamicCacheDescriptor(cfg, IgniteUuid.randomUuid()); desc.locallyConfigured(true); - registeredCaches.put(cfg.getName(), desc); + registeredCaches.put(maskNull(cfg.getName()), desc); ctx.discovery().setCacheFilter( cfg.getName(), @@ -627,25 +626,16 @@ public class GridCacheProcessor extends GridProcessorAdapter { cfg.getNearConfiguration() != null, cfg.getCacheMode() == LOCAL); - sharedCtx.addCacheContext(cacheCtx); - - startSeq.add(cacheCtx.cache()); - - caches.put(cfg.getName(), cacheCtx.cache()); - if (sysCaches.contains(cfg.getName())) - stopSeq.addLast(cacheCtx.cache()); + stopSeq.addLast(cfg.getName()); else - stopSeq.addFirst(cacheCtx.cache()); + stopSeq.addFirst(cfg.getName()); } // Start shared managers. for (GridCacheSharedManager mgr : sharedCtx.managers()) mgr.start(sharedCtx); - for (GridCacheAdapter<?, ?> cache : startSeq) - startCache(cache); - for (Map.Entry<String, GridCacheAdapter<?, ?>> e : caches.entrySet()) { GridCacheAdapter cache = e.getValue(); @@ -654,25 +644,8 @@ public class GridCacheProcessor extends GridProcessorAdapter { jCacheProxies.put(e.getKey(), new IgniteCacheProxy(cache.context(), cache, null, false)); } - // Internal caches which should not be returned to user. - for (Map.Entry<String, GridCacheAdapter<?, ?>> e : caches.entrySet()) { - GridCacheAdapter cache = e.getValue(); - - if (!sysCaches.contains(e.getKey())) - publicProxies.put(e.getKey(), new GridCacheProxyImpl(cache.context(), cache, null)); - } - transactions = new IgniteTransactionsImpl(sharedCtx); - marshallerCache().context().preloader().syncFuture().listen(new CI1<IgniteInternalFuture<?>>() { - @Override public void apply(IgniteInternalFuture<?> f) { - ctx.marshallerContext().onMarshallerCacheReady(ctx); - } - }); - - if (log.isDebugEnabled()) - log.debug("Started cache processor."); - if (log.isDebugEnabled()) log.debug("Started cache processor."); } @@ -706,7 +679,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { IgnitePredicate filter = ccfg.getNodeFilter(); - if (filter.apply(ctx.discovery().localNode()) && !desc.locallyConfigured()) { + if (filter.apply(ctx.discovery().localNode())) { CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(null, ccfg.getName(), ccfg); GridCacheContext ctx = createCache(ccfg, cacheObjCtx); @@ -729,6 +702,20 @@ public class GridCacheProcessor extends GridProcessorAdapter { } } + marshallerCache().context().preloader().syncFuture().listen(new CI1<IgniteInternalFuture<?>>() { + @Override public void apply(IgniteInternalFuture<?> f) { + ctx.marshallerContext().onMarshallerCacheReady(ctx); + } + }); + + // Internal caches which should not be returned to user. + for (Map.Entry<String, GridCacheAdapter<?, ?>> e : caches.entrySet()) { + GridCacheAdapter cache = e.getValue(); + + if (!sysCaches.contains(e.getKey())) + publicProxies.put(e.getKey(), new GridCacheProxyImpl(cache.context(), cache, null)); + } + // Must call onKernalStart on shared managers after creation of fetched caches. for (GridCacheSharedManager<?, ?> mgr : sharedCtx.managers()) mgr.onKernalStart(); @@ -782,8 +769,9 @@ public class GridCacheProcessor extends GridProcessorAdapter { if (ctx.config().isDaemon()) return; - for (GridCacheAdapter<?, ?> cache : stopSeq) - stopCache(cache, cancel); + for (String cacheName : stopSeq) { + stopCache(caches.get(cacheName), cancel); + } List<? extends GridCacheSharedManager<?, ?>> mgrs = sharedCtx.managers(); @@ -805,8 +793,8 @@ public class GridCacheProcessor extends GridProcessorAdapter { if (ctx.config().isDaemon()) return; - for (GridCacheAdapter<?, ?> cache : stopSeq) - onKernalStop(cache, cancel); + for (String cacheName : stopSeq) + onKernalStop(caches.get(cacheName), cancel); List<? extends GridCacheSharedManager<?, ?>> sharedMgrs = sharedCtx.managers(); @@ -1225,7 +1213,12 @@ public class GridCacheProcessor extends GridProcessorAdapter { * @return Collection of started cache names. */ public Collection<String> cacheNames() { - return registeredCaches.keySet(); + return F.viewReadOnly(registeredCaches.keySet(), + new IgniteClosure<String, String>() { + @Override public String apply(String s) { + return unmaskNull(s); + } + }); } /** @@ -1235,7 +1228,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { * @return Cache mode. */ public CacheMode cacheMode(String cacheName) { - DynamicCacheDescriptor desc = registeredCaches.get(cacheName); + DynamicCacheDescriptor desc = registeredCaches.get(maskNull(cacheName)); return desc != null ? desc.cacheConfiguration().getCacheMode() : null; } @@ -1246,7 +1239,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { */ @SuppressWarnings("IfMayBeConditional") public boolean dynamicCacheRegistered(DynamicCacheChangeRequest req) { - DynamicCacheDescriptor desc = registeredCaches.get(req.cacheName()); + DynamicCacheDescriptor desc = registeredCaches.get(maskNull(req.cacheName())); if (desc != null && desc.deploymentId().equals(req.deploymentId())) { if (req.isStart() || req.isClientStart()) @@ -1351,13 +1344,15 @@ public class GridCacheProcessor extends GridProcessorAdapter { jCacheProxies.put(cache.name(), new IgniteCacheProxy(cache.context(), cache, null, false)); } else { - DynamicCacheDescriptor desc = registeredCaches.get(req.cacheName()); + String masked = maskNull(req.cacheName()); + + DynamicCacheDescriptor desc = registeredCaches.get(masked); if (desc != null && desc.cancelled() && desc.deploymentId().equals(req.deploymentId())) - registeredCaches.remove(req.cacheName(), desc); + registeredCaches.remove(masked, desc); } - DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(req.cacheName()); + DynamicCacheStartFuture fut = (DynamicCacheStartFuture)pendingFuts.get(maskNull(req.cacheName())); assert req.deploymentId() != null; assert fut == null || fut.deploymentId != null; @@ -1421,7 +1416,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { DynamicCacheChangeBatch batch = (DynamicCacheChangeBatch)data; for (DynamicCacheChangeRequest req : batch.requests()) { - DynamicCacheDescriptor existing = registeredCaches.get(req.cacheName()); + DynamicCacheDescriptor existing = registeredCaches.get(maskNull(req.cacheName())); try { if (req.isStart()) { @@ -1442,7 +1437,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { } } else - registeredCaches.put(req.cacheName(), new DynamicCacheDescriptor( + registeredCaches.put(maskNull(req.cacheName()), new DynamicCacheDescriptor( ccfg, req.deploymentId())); } @@ -1460,21 +1455,54 @@ public class GridCacheProcessor extends GridProcessorAdapter { * @param ccfg Cache configuration. * @return Future that will be completed when cache is deployed. */ - public IgniteInternalFuture<?> dynamicStartCache(CacheConfiguration ccfg) { - try { - CacheConfiguration cfg = new CacheConfiguration(ccfg); + public IgniteInternalFuture<?> dynamicStartCache( + @Nullable CacheConfiguration ccfg, + @Nullable NearCacheConfiguration nearCfg + ) { + Collection<DynamicCacheChangeRequest> reqs = new ArrayList<>(2); - CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(null, ccfg.getName(), ccfg); + if (ccfg != null) { + try { + CacheConfiguration cfg = new CacheConfiguration(ccfg); - initialize(cfg, cacheObjCtx); + CacheObjectContext cacheObjCtx = ctx.cacheObjects().contextForCache(null, ccfg.getName(), ccfg); + + initialize(cfg, cacheObjCtx); - DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(cfg); + DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(cfg); - return F.first(initiateCacheChanges(F.asList(req))); + reqs.add(req); + } + catch (IgniteCheckedException e) { + return new GridFinishedFuture(e); + } } - catch (IgniteCheckedException e) { - return new GridFinishedFuture(e); + + if (nearCfg != null) { + if (ccfg == null) { + DynamicCacheDescriptor desc = registeredCaches.get(maskNull(nearCfg.getName())); + + if (desc != null && !desc.cancelled()) + ccfg = desc.cacheConfiguration(); + } + + if (ccfg == null) + return new GridFinishedFuture<>(new IgniteCheckedException("Failed to start near cache " + + "(a cache with the given name is not started): " + nearCfg.getName())); + + DynamicCacheChangeRequest req = new DynamicCacheChangeRequest(ctx.localNodeId(), ccfg, nearCfg); + + reqs.add(req); } + + GridCompoundFuture fut = new GridCompoundFuture(); + + for (DynamicCacheStartFuture startFut : initiateCacheChanges(reqs)) + fut.add(startFut); + + fut.markInitialized(); + + return fut; } /** @@ -1498,29 +1526,14 @@ public class GridCacheProcessor extends GridProcessorAdapter { DynamicCacheStartFuture fut = new DynamicCacheStartFuture(req.cacheName(), req.deploymentId()); try { - for (CacheConfiguration ccfg0 : ctx.config().getCacheConfiguration()) { - if (ccfg0.getName().equals(req.cacheName())) { - Exception ex = new IgniteCheckedException("Failed to " + - (req.isStart() ? "start" : "stop") + " cache " + - "(a cache with the same name is manually configured): " + ccfg0.getName()); - - fut.onDone(ex); - - break; - } - } - - if (fut.isDone()) - continue; - if (req.isStart()) { - if (registeredCaches.containsKey(req.cacheName())) { + if (registeredCaches.containsKey(maskNull(req.cacheName()))) { fut.onDone(new IgniteCheckedException("Failed to start cache " + "(a cache with the same name is already started): " + req.cacheName())); } } else if (!req.isClientStart()) { - DynamicCacheDescriptor desc = registeredCaches.get(req.cacheName()); + DynamicCacheDescriptor desc = registeredCaches.get(maskNull(req.cacheName())); if (desc == null) // No-op. @@ -1539,7 +1552,8 @@ public class GridCacheProcessor extends GridProcessorAdapter { if (fut.isDone()) continue; - DynamicCacheStartFuture old = (DynamicCacheStartFuture)pendingFuts.putIfAbsent(req.cacheName(), fut); + DynamicCacheStartFuture old = (DynamicCacheStartFuture)pendingFuts.putIfAbsent( + maskNull(req.cacheName()), fut); if (old != null) { if (req.isStart()) { @@ -1578,7 +1592,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { */ private void onCacheChangeRequested(DynamicCacheChangeBatch batch) { for (DynamicCacheChangeRequest req : batch.requests()) { - DynamicCacheDescriptor desc = registeredCaches.get(req.cacheName()); + DynamicCacheDescriptor desc = registeredCaches.get(maskNull(req.cacheName())); if (req.isStart()) { CacheConfiguration ccfg = req.startCacheConfiguration(); @@ -1586,7 +1600,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { // Check if cache with the same name was concurrently started form different node. if (desc != null) { // If local node initiated start, fail the start future. - DynamicCacheStartFuture startFut = (DynamicCacheStartFuture)pendingFuts.get(ccfg.getName()); + DynamicCacheStartFuture startFut = (DynamicCacheStartFuture)pendingFuts.get(maskNull(ccfg.getName())); if (startFut != null && startFut.deploymentId().equals(req.deploymentId())) { startFut.onDone(new IgniteCheckedException("Failed to start cache " + @@ -1598,7 +1612,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { DynamicCacheDescriptor startDesc = new DynamicCacheDescriptor(ccfg, req.deploymentId()); - DynamicCacheDescriptor old = registeredCaches.put(ccfg.getName(), startDesc); + DynamicCacheDescriptor old = registeredCaches.put(maskNull(ccfg.getName()), startDesc); ctx.discovery().setCacheFilter( ccfg.getName(), @@ -1622,7 +1636,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { else { if (desc == null) { // If local node initiated start, fail the start future. - DynamicCacheStartFuture changeFut = (DynamicCacheStartFuture)pendingFuts.get(req.cacheName()); + DynamicCacheStartFuture changeFut = (DynamicCacheStartFuture)pendingFuts.get(maskNull(req.cacheName())); if (changeFut != null && changeFut.deploymentId().equals(req.deploymentId())) { // No-op. @@ -2072,7 +2086,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { IgniteCache<K,V> cache = (IgniteCache<K, V>)jCacheProxies.get(name); if (cache == null) { - DynamicCacheDescriptor desc = registeredCaches.get(name); + DynamicCacheDescriptor desc = registeredCaches.get(maskNull(name)); if (desc == null || desc.cancelled()) throw new IllegalArgumentException("Cache is not started: " + name); @@ -2298,6 +2312,24 @@ public class GridCacheProcessor extends GridProcessorAdapter { } /** + * @param name Name to mask. + * @return Masked name. + */ + private static String maskNull(String name) { + return name == null ? NULL_NAME : name; + } + + /** + * @param name Name to unmask. + * @return Unmasked name. + */ + @SuppressWarnings("StringEquality") + private static String unmaskNull(String name) { + // Intentional identity equality. + return name == NULL_NAME ? null : name; + } + + /** * */ @SuppressWarnings("ExternalizableWithoutPublicNoArgConstructor") @@ -2325,7 +2357,7 @@ public class GridCacheProcessor extends GridProcessorAdapter { /** {@inheritDoc} */ @Override public boolean onDone(@Nullable Object res, @Nullable Throwable err) { if (super.onDone(res, err)) { - pendingFuts.remove(cacheName, this); + pendingFuts.remove(maskNull(cacheName), this); return true; } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java index cb889f2..08045bc 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/dht/colocated/GridDhtColocatedLockFuture.java @@ -21,7 +21,6 @@ import org.apache.ignite.*; import org.apache.ignite.cluster.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.cluster.*; -import org.apache.ignite.internal.managers.discovery.*; import org.apache.ignite.internal.processors.affinity.*; import org.apache.ignite.internal.processors.cache.*; import org.apache.ignite.internal.processors.cache.distributed.*; @@ -99,8 +98,7 @@ public final class GridDhtColocatedLockFuture<K, V> extends GridCompoundIdentity private GridNearTxLocal tx; /** Topology snapshot to operate on. */ - private AtomicReference<GridDiscoveryTopologySnapshot> topSnapshot = - new AtomicReference<>(); + private AtomicReference<AffinityTopologyVersion> topVer = new AtomicReference<>(); /** Map of current values. */ private Map<KeyCacheObject, IgniteBiTuple<GridCacheVersion, CacheObject>> valMap; @@ -286,7 +284,7 @@ public final class GridDhtColocatedLockFuture<K, V> extends GridCompoundIdentity false, false); - cand.topologyVersion(new AffinityTopologyVersion(topSnapshot.get().topologyVersion())); + cand.topologyVersion(new AffinityTopologyVersion(topVer.get().topologyVersion())); } } else { @@ -305,12 +303,12 @@ public final class GridDhtColocatedLockFuture<K, V> extends GridCompoundIdentity false, false); - cand.topologyVersion(new AffinityTopologyVersion(topSnapshot.get().topologyVersion())); + cand.topologyVersion(new AffinityTopologyVersion(topVer.get().topologyVersion())); } else cand = cand.reenter(); - cctx.mvcc().addExplicitLock(threadId, cand, topSnapshot.get()); + cctx.mvcc().addExplicitLock(threadId, cand, topVer.get()); } return cand; @@ -519,12 +517,12 @@ public final class GridDhtColocatedLockFuture<K, V> extends GridCompoundIdentity */ void map() { // Obtain the topology version to use. - GridDiscoveryTopologySnapshot snapshot = tx != null ? tx.topologySnapshot() : - cctx.mvcc().lastExplicitLockTopologySnapshot(threadId); + AffinityTopologyVersion topVer = tx != null ? tx.topologyVersionSnapshot() : + cctx.mvcc().lastExplicitLockTopologyVersion(threadId); - if (snapshot != null) { + if (topVer != null) { // Continue mapping on the same topology version as it was before. - topSnapshot.compareAndSet(null, snapshot); + this.topVer.compareAndSet(null, topVer); map(keys); @@ -543,40 +541,33 @@ public final class GridDhtColocatedLockFuture<K, V> extends GridCompoundIdentity */ private void mapOnTopology() { // We must acquire topology snapshot from the topology version future. - try { - cctx.topology().readLock(); + cctx.topology().readLock(); - try { - GridDhtTopologyFuture fut = cctx.topologyVersionFuture(); + try { + GridDhtTopologyFuture fut = cctx.topologyVersionFuture(); - if (fut.isDone()) { - GridDiscoveryTopologySnapshot snapshot = fut.topologySnapshot(); + if (fut.isDone()) { + AffinityTopologyVersion topVer = fut.topologyVersion(); - if (tx != null) { - tx.topologyVersion(new AffinityTopologyVersion(snapshot.topologyVersion())); - tx.topologySnapshot(snapshot); - } + if (tx != null) + tx.topologyVersion(topVer); - topSnapshot.compareAndSet(null, snapshot); + this.topVer.compareAndSet(null, topVer); - map(keys); + map(keys); - markInitialized(); - } - else { - fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() { - @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) { - mapOnTopology(); - } - }); - } + markInitialized(); } - finally { - cctx.topology().readUnlock(); + else { + fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() { + @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) { + mapOnTopology(); + } + }); } } - catch (IgniteCheckedException e) { - onDone(e); + finally { + cctx.topology().readUnlock(); } } @@ -589,11 +580,9 @@ public final class GridDhtColocatedLockFuture<K, V> extends GridCompoundIdentity */ private void map(Collection<KeyCacheObject> keys) { try { - GridDiscoveryTopologySnapshot snapshot = topSnapshot.get(); - - assert snapshot != null; + AffinityTopologyVersion topVer = this.topVer.get(); - final AffinityTopologyVersion topVer = new AffinityTopologyVersion(snapshot.topologyVersion()); + assert topVer != null; assert topVer.topologyVersion() > 0; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java index df2ffde..1dfeb45 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearLockFuture.java @@ -21,7 +21,6 @@ import org.apache.ignite.*; import org.apache.ignite.cluster.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.cluster.*; -import org.apache.ignite.internal.managers.discovery.*; import org.apache.ignite.internal.processors.affinity.*; import org.apache.ignite.internal.processors.cache.*; import org.apache.ignite.internal.processors.cache.distributed.*; @@ -101,7 +100,7 @@ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<B private GridNearTxLocal tx; /** Topology snapshot to operate on. */ - private AtomicReference<GridDiscoveryTopologySnapshot> topSnapshot = + private AtomicReference<AffinityTopologyVersion> topVer = new AtomicReference<>(); /** Map of current values. */ @@ -650,12 +649,12 @@ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<B */ void map() { // Obtain the topology version to use. - GridDiscoveryTopologySnapshot snapshot = tx != null ? tx.topologySnapshot() : - cctx.mvcc().lastExplicitLockTopologySnapshot(Thread.currentThread().getId()); + AffinityTopologyVersion topVer = tx != null ? tx.topologyVersionSnapshot() : + cctx.mvcc().lastExplicitLockTopologyVersion(Thread.currentThread().getId()); - if (snapshot != null) { + if (topVer != null) { // Continue mapping on the same topology version as it was before. - topSnapshot.compareAndSet(null, snapshot); + this.topVer.compareAndSet(null, topVer); map(keys); @@ -674,40 +673,33 @@ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<B */ void mapOnTopology() { // We must acquire topology snapshot from the topology version future. - try { - cctx.topology().readLock(); + cctx.topology().readLock(); - try { - GridDhtTopologyFuture fut = cctx.topologyVersionFuture(); + try { + GridDhtTopologyFuture fut = cctx.topologyVersionFuture(); - if (fut.isDone()) { - GridDiscoveryTopologySnapshot snapshot = fut.topologySnapshot(); + if (fut.isDone()) { + AffinityTopologyVersion topVer = fut.topologyVersion(); - if (tx != null) { - tx.topologyVersion(new AffinityTopologyVersion(snapshot.topologyVersion())); - tx.topologySnapshot(snapshot); - } + if (tx != null) + tx.topologyVersion(topVer); - topSnapshot.compareAndSet(null, snapshot); + this.topVer.compareAndSet(null, topVer); - map(keys); + map(keys); - markInitialized(); - } - else { - fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() { - @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) { - mapOnTopology(); - } - }); - } + markInitialized(); } - finally { - cctx.topology().readUnlock(); + else { + fut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() { + @Override public void apply(IgniteInternalFuture<AffinityTopologyVersion> t) { + mapOnTopology(); + } + }); } } - catch (IgniteCheckedException e) { - onDone(e); + finally { + cctx.topology().readUnlock(); } } @@ -720,11 +712,9 @@ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<B */ private void map(Iterable<KeyCacheObject> keys) { try { - GridDiscoveryTopologySnapshot snapshot = topSnapshot.get(); - - assert snapshot != null; + AffinityTopologyVersion topVer = this.topVer.get(); - AffinityTopologyVersion topVer = new AffinityTopologyVersion(snapshot.topologyVersion()); + assert topVer != null; assert topVer.topologyVersion() > 0; @@ -811,7 +801,7 @@ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<B if (cand != null) { if (tx == null && !cand.reentry()) - cctx.mvcc().addExplicitLock(threadId, cand, snapshot); + cctx.mvcc().addExplicitLock(threadId, cand, topVer); IgniteBiTuple<GridCacheVersion, CacheObject> val = entry.versionedValue(); @@ -1328,7 +1318,7 @@ public final class GridNearLockFuture<K, V> extends GridCompoundIdentityFuture<B int i = 0; - AffinityTopologyVersion topVer = new AffinityTopologyVersion(topSnapshot.get().topologyVersion()); + AffinityTopologyVersion topVer = GridNearLockFuture.this.topVer.get(); for (KeyCacheObject k : keys) { while (true) { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java index d5ff286..4776b23 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxLocal.java @@ -50,14 +50,11 @@ import static org.apache.ignite.transactions.TransactionState.*; /** * Replicated user transaction. */ +@SuppressWarnings("unchecked") public class GridNearTxLocal extends GridDhtTxLocalAdapter { /** */ private static final long serialVersionUID = 0L; - /** Topology snapshot on which this tx was started. */ - private final AtomicReference<GridDiscoveryTopologySnapshot> topSnapshot = - new AtomicReference<>(); - /** DHT mappings. */ private ConcurrentMap<UUID, GridDistributedTxMapping> mappings = new ConcurrentHashMap8<>(); @@ -597,23 +594,6 @@ public class GridNearTxLocal extends GridDhtTxLocalAdapter { } } - /** - * @return Topology snapshot on which this tx was started. - */ - public GridDiscoveryTopologySnapshot topologySnapshot() { - return topSnapshot.get(); - } - - /** - * Sets topology snapshot on which this tx was started. - * - * @param topSnapshot Topology snapshot. - * @return {@code True} if topology snapshot was set by this call. - */ - public boolean topologySnapshot(GridDiscoveryTopologySnapshot topSnapshot) { - return this.topSnapshot.compareAndSet(null, topSnapshot); - } - /** {@inheritDoc} */ @SuppressWarnings({"CatchGenericClass", "ThrowableInstanceNeverThrown"}) @Override public boolean finish(boolean commit) throws IgniteCheckedException { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java index c269283..5d4daf4 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/distributed/near/GridNearTxPrepareFuture.java @@ -21,7 +21,6 @@ import org.apache.ignite.*; import org.apache.ignite.cluster.*; import org.apache.ignite.internal.*; import org.apache.ignite.internal.cluster.*; -import org.apache.ignite.internal.managers.discovery.*; import org.apache.ignite.internal.processors.affinity.*; import org.apache.ignite.internal.processors.cache.*; import org.apache.ignite.internal.processors.cache.distributed.*; @@ -330,10 +329,7 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut return; } - GridDiscoveryTopologySnapshot snapshot = topFut.topologySnapshot(); - tx.topologyVersion(topFut.topologyVersion()); - tx.topologySnapshot(snapshot); // Make sure to add future before calling prepare. cctx.mvcc().addFuture(this); @@ -343,17 +339,6 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut catch (TransactionTimeoutException | TransactionOptimisticException e) { onError(cctx.localNodeId(), null, e); } - catch (IgniteCheckedException e) { - tx.setRollbackOnly(); - - String msg = "Failed to prepare transaction (will attempt rollback): " + this; - - U.error(log, msg, e); - - tx.rollbackAsync(); - - onError(null, null, new IgniteTxRollbackCheckedException(msg, e)); - } } else { topFut.listen(new CI1<IgniteInternalFuture<AffinityTopologyVersion>>() { @@ -455,10 +440,6 @@ public final class GridNearTxPrepareFuture<K, V> extends GridCompoundIdentityFut ) throws IgniteCheckedException { assert tx.optimistic(); - GridDiscoveryTopologySnapshot snapshot = tx.topologySnapshot(); - - assert snapshot != null; - AffinityTopologyVersion topVer = tx.topologyVersion(); assert topVer.topologyVersion() > 0; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java index a4c1fdb..8dcfcb1 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteInternalTx.java @@ -254,6 +254,11 @@ public interface IgniteInternalTx extends AutoCloseable, GridTimeoutObject { public AffinityTopologyVersion topologyVersion(); /** + * @return Topology version snapshot. + */ + public AffinityTopologyVersion topologyVersionSnapshot(); + + /** * @return Flag indicating whether transaction is implicit with only one key. */ public boolean implicitSingle(); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java index 3993527..acd3202 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/transactions/IgniteTxAdapter.java @@ -45,7 +45,6 @@ import java.util.concurrent.locks.*; import static org.apache.ignite.events.EventType.*; import static org.apache.ignite.internal.processors.cache.GridCacheOperation.*; -import static org.apache.ignite.internal.processors.cache.GridCacheUtils.*; import static org.apache.ignite.transactions.TransactionConcurrency.*; import static org.apache.ignite.transactions.TransactionIsolation.*; import static org.apache.ignite.transactions.TransactionState.*; @@ -518,6 +517,13 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter } /** {@inheritDoc} */ + @Override public AffinityTopologyVersion topologyVersionSnapshot() { + AffinityTopologyVersion ret = topVer.get(); + + return AffinityTopologyVersion.NONE.equals(ret) ? null : ret; + } + + /** {@inheritDoc} */ @Override public AffinityTopologyVersion topologyVersion(AffinityTopologyVersion topVer) { this.topVer.compareAndSet(AffinityTopologyVersion.NONE, topVer); @@ -1745,6 +1751,11 @@ public abstract class IgniteTxAdapter extends GridMetadataAwareAdapter } /** {@inheritDoc} */ + @Override public AffinityTopologyVersion topologyVersionSnapshot() { + throw new IllegalStateException("Deserialized transaction can only be used as read-only."); + } + + /** {@inheritDoc} */ @Override public boolean implicitSingle() { throw new IllegalStateException("Deserialized transaction can only be used as read-only."); } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionManager.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionManager.java index d5b3db7..c776361 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionManager.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/cache/version/GridCacheVersionManager.java @@ -89,8 +89,6 @@ public class GridCacheVersionManager extends GridCacheSharedManagerAdapter { @Override public void start0() throws IgniteCheckedException { txSerEnabled = cctx.gridConfig().getTransactionConfiguration().isTxSerializableEnabled(); - dataCenterId = cctx.dataCenterId(); - last = new GridCacheVersion(0, 0, order.get(), 0, dataCenterId); cctx.gridEvents().addLocalEventListener(discoLsnr, EVT_NODE_METRICS_UPDATED); @@ -108,6 +106,17 @@ public class GridCacheVersionManager extends GridCacheSharedManagerAdapter { } /** + * Sets data center ID. + * + * @param dataCenterId Data center ID. + */ + public void dataCenterId(byte dataCenterId) { + this.dataCenterId = dataCenterId; + + last = new GridCacheVersion(0, 0, order.get(), 0, dataCenterId); + } + + /** * @param nodeId Node ID. * @param ver Remote version. */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java index d5641df..49081ae 100644 --- a/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java +++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/datastructures/DataStructuresProcessor.java @@ -105,7 +105,7 @@ public final class DataStructuresProcessor extends GridProcessorAdapter { /** {@inheritDoc} */ @SuppressWarnings("unchecked") - @Override public void start() { + @Override public void onKernalStart() { if (ctx.config().isDaemon()) return; http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java ---------------------------------------------------------------------- diff --git a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java index 5061353..6193aec 100644 --- a/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java +++ b/modules/core/src/main/java/org/apache/ignite/spi/discovery/tcp/TcpDiscoverySpi.java @@ -1458,6 +1458,8 @@ public class TcpDiscoverySpi extends TcpDiscoverySpiAdapter implements TcpDiscov } } catch (IgniteSpiException e) { + e.printStackTrace(); + ex = e; } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFlagsTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFlagsTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFlagsTest.java index ee35555..72a8cfe 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFlagsTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractFlagsTest.java @@ -49,6 +49,8 @@ public abstract class GridCacheAbstractFlagsTest extends GridCacheAbstractSelfTe if (cacheMode() == CacheMode.PARTITIONED) c.setBackups(1); + c.setCacheStoreFactory(null); + return c; } @@ -57,11 +59,6 @@ public abstract class GridCacheAbstractFlagsTest extends GridCacheAbstractSelfTe return false; } - /** {@inheritDoc} */ - @Override protected CacheStore<?, ?> cacheStore() { - return null; - } - /** * Tests SYNC_COMMIT cache flag. * http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java index cfe4fdc..e4e2fc8 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheAbstractSelfTest.java @@ -43,7 +43,6 @@ import java.util.*; import java.util.concurrent.atomic.*; import static org.apache.ignite.cache.CacheAtomicityMode.*; -import static org.apache.ignite.cache.CacheDistributionMode.*; import static org.apache.ignite.cache.CacheMemoryMode.*; import static org.apache.ignite.cache.CacheMode.*; import static org.apache.ignite.cache.CacheWriteSynchronizationMode.*; @@ -77,7 +76,7 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest { assert cnt >= 1 : "At least one grid must be started"; - startGridsMultiThreaded(cnt); + startGrids(cnt); } /** {@inheritDoc} */ @@ -104,6 +103,8 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest { } for (int i = 0; i < gridCount(); i++) { + info("Checking grid: " + i); + while (true) { try { final int fi = i; @@ -241,7 +242,7 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest { CacheStore<?, ?> store = cacheStore(); if (store != null) { - cfg.setCacheStoreFactory(new FactoryBuilder.SingletonFactory(store)); + cfg.setCacheStoreFactory(new TestStoreFactory()); cfg.setReadThrough(true); cfg.setWriteThrough(true); cfg.setLoadPreviousValue(true); @@ -290,7 +291,7 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest { /** * @return Write through storage emulator. */ - protected CacheStore<?, ?> cacheStore() { + protected static CacheStore<?, ?> cacheStore() { return new CacheStoreAdapter<Object, Object>() { @Override public void loadCache(IgniteBiInClosure<Object, Object> clo, Object... args) { @@ -596,4 +597,13 @@ public abstract class GridCacheAbstractSelfTest extends GridCommonAbstractTest { return sum; } } + + /** + * Serializable factory. + */ + private static class TestStoreFactory implements Factory<CacheStore> { + @Override public CacheStore create() { + return cacheStore(); + } + } } http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheMultinodeUpdateAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheMultinodeUpdateAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheMultinodeUpdateAbstractSelfTest.java index de1769c..886d8fa 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheMultinodeUpdateAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/GridCacheMultinodeUpdateAbstractSelfTest.java @@ -19,9 +19,8 @@ package org.apache.ignite.internal.processors.cache; import org.apache.ignite.*; import org.apache.ignite.cache.*; -import org.apache.ignite.cache.store.*; +import org.apache.ignite.configuration.*; import org.apache.ignite.testframework.*; -import org.jetbrains.annotations.*; import javax.cache.processor.*; import java.io.*; @@ -44,11 +43,6 @@ public abstract class GridCacheMultinodeUpdateAbstractSelfTest extends GridCache } /** {@inheritDoc} */ - @Nullable @Override protected CacheStore<?, ?> cacheStore() { - return null; - } - - /** {@inheritDoc} */ @Override protected CacheMode cacheMode() { return PARTITIONED; } @@ -58,6 +52,14 @@ public abstract class GridCacheMultinodeUpdateAbstractSelfTest extends GridCache return 3 * 60_000; } + @Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception { + CacheConfiguration ccfg = super.cacheConfiguration(gridName); + + ccfg.setCacheStoreFactory(null); + + return ccfg; + } + /** * @throws Exception If failed. */ http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java index 6efc2a8..6ea7cb4 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/IgniteDynamicCacheStartSelfTest.java @@ -105,7 +105,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { ccfg.setName(DYNAMIC_CACHE_NAME); - futs.add(kernal.context().cache().dynamicStartCache(ccfg)); + futs.add(kernal.context().cache().dynamicStartCache(ccfg, null)); return null; } @@ -166,7 +166,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { IgniteKernal kernal = (IgniteKernal)grid(ThreadLocalRandom.current().nextInt(nodeCount())); - futs.add(kernal.context().cache().dynamicStartCache(ccfg)); + futs.add(kernal.context().cache().dynamicStartCache(ccfg, null)); return null; } @@ -239,7 +239,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { ccfg.setName(DYNAMIC_CACHE_NAME); - kernal.context().cache().dynamicStartCache(ccfg).get(); + kernal.context().cache().dynamicStartCache(ccfg, null).get(); for (int g = 0; g < nodeCount(); g++) { IgniteKernal kernal0 = (IgniteKernal)grid(g); @@ -298,7 +298,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { ccfg.setName(DYNAMIC_CACHE_NAME); - kernal.context().cache().dynamicStartCache(ccfg).get(); + kernal.context().cache().dynamicStartCache(ccfg, null).get(); info(">>>>>>> Deployed dynamic cache"); @@ -359,7 +359,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { ccfg.setNodeFilter(NODE_FILTER); - kernal.context().cache().dynamicStartCache(ccfg).get(); + kernal.context().cache().dynamicStartCache(ccfg, null).get(); startGrid(nodeCount() + 1); @@ -424,7 +424,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { ccfg.setNodeFilter(NODE_FILTER); - return kernal.context().cache().dynamicStartCache(ccfg).get(); + return kernal.context().cache().dynamicStartCache(ccfg, null).get(); } }, IgniteCheckedException.class, null); } @@ -447,7 +447,7 @@ public class IgniteDynamicCacheStartSelfTest extends GridCommonAbstractTest { ccfg.setNodeFilter(NODE_FILTER); - kernal.context().cache().dynamicStartCache(ccfg).get(); + kernal.context().cache().dynamicStartCache(ccfg, null).get(); GridTestUtils.assertThrows(log, new Callable<Object>() { @Override public Object call() throws Exception { http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java index 6f0b923..4911a49 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/GridCacheClientModesAbstractSelfTest.java @@ -54,11 +54,6 @@ public abstract class GridCacheClientModesAbstractSelfTest extends GridCacheAbst } /** {@inheritDoc} */ - @Override protected CacheStore<?, ?> cacheStore() { - return null; - } - - /** {@inheritDoc} */ @Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception { CacheConfiguration cfg = super.cacheConfiguration(gridName); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOnlyMultiNodeFullApiSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOnlyMultiNodeFullApiSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOnlyMultiNodeFullApiSelfTest.java index 4b425fe..458d86f 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOnlyMultiNodeFullApiSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/distributed/near/GridCacheNearOnlyMultiNodeFullApiSelfTest.java @@ -22,7 +22,6 @@ import org.apache.ignite.cache.*; import org.apache.ignite.cluster.*; import org.apache.ignite.configuration.*; import org.apache.ignite.events.*; -import org.apache.ignite.internal.*; import org.apache.ignite.internal.processors.cache.*; import org.apache.ignite.internal.util.lang.*; import org.apache.ignite.internal.util.typedef.*; @@ -58,16 +57,22 @@ public class GridCacheNearOnlyMultiNodeFullApiSelfTest extends GridCachePartitio super.beforeTestsStarted(); } - /** {@inheritDoc} */ - @Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception { - CacheConfiguration cfg = super.cacheConfiguration(gridName); + @Override protected IgniteConfiguration getConfiguration(String gridName) throws Exception { + IgniteConfiguration cfg = super.getConfiguration(gridName); if (cnt.getAndIncrement() == 0) { info("Use grid '" + gridName + "' as near-only."); - return null; + cfg.setClientMode(true); } + return cfg; + } + + /** {@inheritDoc} */ + @Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception { + CacheConfiguration cfg = super.cacheConfiguration(gridName); + cfg.setWriteSynchronizationMode(FULL_SYNC); cfg.setAtomicWriteOrderMode(PRIMARY); @@ -77,7 +82,9 @@ public class GridCacheNearOnlyMultiNodeFullApiSelfTest extends GridCachePartitio /** {@inheritDoc} */ @Override protected void beforeTest() throws Exception { for (int i = 0; i < gridCount(); i++) { - if (!cache(i).affinityNode()) { + if (ignite(i).configuration().isClientMode()) { + ignite(i).createCache(new NearCacheConfiguration<>()); + nearIdx = i; break; @@ -95,15 +102,6 @@ public class GridCacheNearOnlyMultiNodeFullApiSelfTest extends GridCachePartitio @Override public Collection<ClusterNode> affinityNodes() { info("Near node ID: " + grid(nearIdx).localNode().id()); - for (int i = 0; i < gridCount(); i++) { - ClusterNode node = grid(i).localNode(); - - GridCacheAttributes[] nodeAttrs = node.attribute(IgniteNodeAttributes.ATTR_CACHE); - - info("Cache attributes for node [nodeId=" + node.id() + ", attrs=" + - Arrays.asList(nodeAttrs) + ']'); - } - return F.view(super.affinityNodes(), new P1<ClusterNode>() { @Override public boolean apply(ClusterNode n) { return !F.eq(G.ignite(n.id()).name(), grid(nearIdx).name()); http://git-wip-us.apache.org/repos/asf/incubator-ignite/blob/65d22d58/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheAtomicLocalMetricsNoStoreSelfTest.java ---------------------------------------------------------------------- diff --git a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheAtomicLocalMetricsNoStoreSelfTest.java b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheAtomicLocalMetricsNoStoreSelfTest.java index 3a4ac6c..e2f953a 100644 --- a/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheAtomicLocalMetricsNoStoreSelfTest.java +++ b/modules/core/src/test/java/org/apache/ignite/internal/processors/cache/local/GridCacheAtomicLocalMetricsNoStoreSelfTest.java @@ -17,13 +17,18 @@ package org.apache.ignite.internal.processors.cache.local; -import org.apache.ignite.cache.store.*; +import org.apache.ignite.configuration.*; /** * Local atomic cache metrics test. */ public class GridCacheAtomicLocalMetricsNoStoreSelfTest extends GridCacheAtomicLocalMetricsSelfTest { - @Override protected CacheStore<?, ?> cacheStore() { - return null; // Test cache without store. + /** {@inheritDoc} */ + @Override protected CacheConfiguration cacheConfiguration(String gridName) throws Exception { + CacheConfiguration ccfg = super.cacheConfiguration(gridName); + + ccfg.setCacheStoreFactory(null); + + return ccfg; } }
