You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by sb...@apache.org on 2018/03/21 04:07:14 UTC
[04/11] ignite git commit: IGNITE-7794 - Share persisted marshaller
mappings when connecting - Fixes #3620.
IGNITE-7794 - Share persisted marshaller mappings when connecting - Fixes #3620.
Signed-off-by: Valentin Kulichenko <va...@gmail.com>
Project: http://git-wip-us.apache.org/repos/asf/ignite/repo
Commit: http://git-wip-us.apache.org/repos/asf/ignite/commit/c4ccf55d
Tree: http://git-wip-us.apache.org/repos/asf/ignite/tree/c4ccf55d
Diff: http://git-wip-us.apache.org/repos/asf/ignite/diff/c4ccf55d
Branch: refs/heads/ignite-zk
Commit: c4ccf55dd6eb22102c56e211812eb4ccc206f4a7
Parents: 5eb54e2
Author: Denis Mekhanikov <dm...@gmail.com>
Authored: Mon Mar 19 19:19:35 2018 -0700
Committer: Valentin Kulichenko <va...@gmail.com>
Committed: Mon Mar 19 19:19:35 2018 -0700
----------------------------------------------------------------------
.../ignite/internal/MarshallerContextImpl.java | 15 +-
.../GridMarshallerMappingProcessor.java | 31 +++-
.../GridMarshallerMappingConsistencyTest.java | 179 +++++++++++++++++++
3 files changed, 219 insertions(+), 6 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/ignite/blob/c4ccf55d/modules/core/src/main/java/org/apache/ignite/internal/MarshallerContextImpl.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/MarshallerContextImpl.java b/modules/core/src/main/java/org/apache/ignite/internal/MarshallerContextImpl.java
index dae95ae..9bad1ea 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/MarshallerContextImpl.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/MarshallerContextImpl.java
@@ -185,12 +185,21 @@ public class MarshallerContextImpl implements MarshallerContext {
/**
* @param platformId Platform id.
* @param marshallerMappings All marshaller mappings for given platformId.
+ * @throws IgniteCheckedException In case of failure to process incoming marshaller mappings.
*/
- public void onMappingDataReceived(byte platformId, Map<Integer, MappedName> marshallerMappings) {
+ public void onMappingDataReceived(byte platformId, Map<Integer, MappedName> marshallerMappings)
+ throws IgniteCheckedException
+ {
ConcurrentMap<Integer, MappedName> platformCache = getCacheFor(platformId);
- for (Map.Entry<Integer, MappedName> e : marshallerMappings.entrySet())
- platformCache.put(e.getKey(), new MappedName(e.getValue().className(), true));
+ for (Map.Entry<Integer, MappedName> e : marshallerMappings.entrySet()) {
+ int typeId = e.getKey();
+ String clsName = e.getValue().className();
+
+ platformCache.put(typeId, new MappedName(clsName, true));
+
+ fileStore.mergeAndWriteMapping(platformId, typeId, clsName);
+ }
}
/**
http://git-wip-us.apache.org/repos/asf/ignite/blob/c4ccf55d/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
----------------------------------------------------------------------
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
index 85b9eb5..81b6f61 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/marshaller/GridMarshallerMappingProcessor.java
@@ -319,21 +319,46 @@ public class GridMarshallerMappingProcessor extends GridProcessorAdapter {
}
/** {@inheritDoc} */
+ @Override public void collectJoiningNodeData(DiscoveryDataBag dataBag) {
+ dataBag.addJoiningNodeData(MARSHALLER_PROC.ordinal(), marshallerCtx.getCachedMappings());
+ }
+
+ /** {@inheritDoc} */
@Override public void collectGridNodeData(DiscoveryDataBag dataBag) {
if (!dataBag.commonDataCollectedFor(MARSHALLER_PROC.ordinal()))
dataBag.addGridCommonData(MARSHALLER_PROC.ordinal(), marshallerCtx.getCachedMappings());
}
/** {@inheritDoc} */
+ @Override public void onJoiningNodeDataReceived(DiscoveryDataBag.JoiningNodeDiscoveryData data) {
+ List<Map<Integer, MappedName>> mappings = (List<Map<Integer, MappedName>>)data.joiningNodeData();
+
+ processIncomingMappings(mappings);
+ }
+
+ /** {@inheritDoc} */
@Override public void onGridDataReceived(GridDiscoveryData data) {
- List<Map<Integer, MappedName>> mappings = (List<Map<Integer, MappedName>>) data.commonData();
+ List<Map<Integer, MappedName>> mappings = (List<Map<Integer, MappedName>>)data.commonData();
+ processIncomingMappings(mappings);
+ }
+
+ /**
+ * @param mappings Incoming marshaller mappings.
+ */
+ private void processIncomingMappings(List<Map<Integer, MappedName>> mappings) {
if (mappings != null) {
for (int i = 0; i < mappings.size(); i++) {
Map<Integer, MappedName> map;
- if ((map = mappings.get(i)) != null)
- marshallerCtx.onMappingDataReceived((byte) i, map);
+ if ((map = mappings.get(i)) != null) {
+ try {
+ marshallerCtx.onMappingDataReceived((byte)i, map);
+ }
+ catch (IgniteCheckedException e) {
+ U.error(log, "Failed to process marshaller mapping data", e);
+ }
+ }
}
}
}
http://git-wip-us.apache.org/repos/asf/ignite/blob/c4ccf55d/modules/core/src/test/java/org/apache/ignite/marshaller/GridMarshallerMappingConsistencyTest.java
----------------------------------------------------------------------
diff --git a/modules/core/src/test/java/org/apache/ignite/marshaller/GridMarshallerMappingConsistencyTest.java b/modules/core/src/test/java/org/apache/ignite/marshaller/GridMarshallerMappingConsistencyTest.java
new file mode 100644
index 0000000..cd8e757
--- /dev/null
+++ b/modules/core/src/test/java/org/apache/ignite/marshaller/GridMarshallerMappingConsistencyTest.java
@@ -0,0 +1,179 @@
+/*
+ * 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.marshaller;
+
+import java.io.File;
+import java.io.IOException;
+import org.apache.commons.io.FileUtils;
+import org.apache.ignite.Ignite;
+import org.apache.ignite.IgniteCache;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.DataRegionConfiguration;
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.TcpDiscoveryIpFinder;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+
+/**
+ *
+ */
+public class GridMarshallerMappingConsistencyTest extends GridCommonAbstractTest {
+ /** Test cache name. */
+ private static final String CACHE_NAME = "cache";
+
+ /** Work directory. */
+ private static final String WORK_DIR = U.getIgniteHome() +
+ File.separatorChar + "work" +
+ File.separatorChar + "test";
+
+ /** Ip finder. */
+ private TcpDiscoveryIpFinder ipFinder = new TcpDiscoveryVmIpFinder(true);
+
+ /** {@inheritDoc} */
+ @Override protected IgniteConfiguration getConfiguration(String igniteInstanceName) throws Exception {
+ IgniteConfiguration igniteCfg = super.getConfiguration(igniteInstanceName);
+
+ igniteCfg.setConsistentId(igniteInstanceName);
+
+ DataRegionConfiguration drCfg = new DataRegionConfiguration();
+ drCfg.setPersistenceEnabled(true);
+
+ DataStorageConfiguration dsCfg = new DataStorageConfiguration();
+ dsCfg.setDefaultDataRegionConfiguration(drCfg);
+
+ igniteCfg.setDataStorageConfiguration(dsCfg);
+
+ igniteCfg.setWorkDirectory(WORK_DIR + File.separator + igniteInstanceName);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+ discoSpi.setIpFinder(ipFinder);
+
+ igniteCfg.setDiscoverySpi(discoSpi);
+
+ return igniteCfg;
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ clearWorkDir();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ stopAllGrids();
+
+ clearWorkDir();
+ }
+
+ /**
+ * @throws IOException If IO error happens.
+ */
+ private void clearWorkDir() throws IOException {
+ FileUtils.deleteDirectory(new File(WORK_DIR));
+ }
+
+ /**
+ * Make a value be rebalanced to a node after the mapping was accepted.
+ * Check, that the mapping is available after restart.
+ *
+ * @throws Exception If failed.
+ */
+ public void testMappingsPersistedOnJoin() throws Exception {
+ Ignite g1 = startGrid(1);
+ Ignite g2 = startGrid(2);
+
+ g1.cluster().active(true);
+
+ CacheConfiguration<Integer, DummyObject> cacheCfg = new CacheConfiguration<>(CACHE_NAME);
+ cacheCfg.setBackups(1);
+
+ IgniteCache<Integer, DummyObject> c1 = g1.getOrCreateCache(cacheCfg);
+ IgniteCache<Integer, DummyObject> c2 = g2.getOrCreateCache(cacheCfg);
+
+ int k = primaryKey(c2);
+
+ stopGrid(2);
+
+ c1.put(k, new DummyObject(k));
+
+ startGrid(2);
+ waitForRebalancing();
+
+ stopAllGrids();
+
+ g2 = startGrid(2);
+
+ g2.cluster().active(true);
+
+ c2 = g2.cache(CACHE_NAME);
+
+ assertEquals(k, c2.get(k).val);
+ }
+
+ /**
+ * Connect a node to a cluster, that already has persisted data and a mapping.
+ * Check, that persisted mappings are distributed to existing nodes.
+ *
+ * @throws Exception If failed.
+ */
+ public void testPersistedMappingsSharedOnJoin() throws Exception {
+ Ignite g1 = startGrid(1);
+ startGrid(2);
+
+ g1.cluster().active(true); // Include second node into baseline topology.
+
+ stopGrid(2);
+
+ IgniteCache<Integer, DummyObject> c1 = g1.getOrCreateCache(CACHE_NAME);
+
+ int k = primaryKey(c1);
+
+ c1.put(k, new DummyObject(k));
+
+ stopAllGrids();
+
+ Ignite g2 = startGrid(2);
+ startGrid(1);
+
+ IgniteCache<Integer, DummyObject> c2 = g2.cache(CACHE_NAME);
+
+ assertEquals(k, c2.get(k).val);
+ }
+
+ /**
+ *
+ */
+ private static class DummyObject {
+ /** */
+ int val;
+
+ /**
+ * @param val Value.
+ */
+ DummyObject(int val) {
+ this.val = val;
+ }
+ }
+}