You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by vo...@apache.org on 2016/09/21 14:52:54 UTC
[02/92] [abbrv] [partial] ignite git commit: Moving classes around.
http://git-wip-us.apache.org/repos/asf/ignite/blob/67b4da70/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/examples/HadoopWordCount2Reducer.java
----------------------------------------------------------------------
diff --git a/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/examples/HadoopWordCount2Reducer.java b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/examples/HadoopWordCount2Reducer.java
new file mode 100644
index 0000000..f24288e
--- /dev/null
+++ b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/examples/HadoopWordCount2Reducer.java
@@ -0,0 +1,113 @@
+/*
+ * 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.processors.hadoop.impl.examples;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configurable;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.IntWritable;
+import org.apache.hadoop.io.Text;
+import org.apache.hadoop.mapreduce.Reducer;
+import org.apache.ignite.internal.processors.hadoop.impl.HadoopErrorSimulator;
+
+/**
+ * Combiner and Reducer phase of WordCount job.
+ */
+public class HadoopWordCount2Reducer extends Reducer<Text, IntWritable, Text, IntWritable> implements Configurable {
+ /** Writable container for writing sum of word counts. */
+ private IntWritable totalWordCnt = new IntWritable();
+
+ /** Flag is to check that mapper was configured before run. */
+ private boolean wasConfigured;
+
+ /** Flag is to check that mapper was set up before run. */
+ private boolean wasSetUp;
+
+ /** {@inheritDoc} */
+ @Override public void reduce(Text key, Iterable<IntWritable> values, Context ctx) throws IOException, InterruptedException {
+ assert wasConfigured : "Reducer should be configured";
+ assert wasSetUp : "Reducer should be set up";
+
+ int wordCnt = 0;
+
+ for (IntWritable value : values)
+ wordCnt += value.get();
+
+ totalWordCnt.set(wordCnt);
+
+ ctx.write(key, totalWordCnt);
+
+ reduceError();
+ }
+
+ /**
+ * Simulates reduce error if needed.
+ */
+ protected void reduceError() throws IOException, InterruptedException {
+ HadoopErrorSimulator.instance().onReduce();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void setup(Context context) throws IOException, InterruptedException {
+ super.setup(context);
+
+ wasSetUp = true;
+
+ setupError();
+ }
+
+ /**
+ * Simulates setup error if needed.
+ */
+ protected void setupError() throws IOException, InterruptedException {
+ HadoopErrorSimulator.instance().onReduceSetup();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void cleanup(Context context) throws IOException, InterruptedException {
+ super.cleanup(context);
+
+ cleanupError();
+ }
+
+ /**
+ * Simulates cleanup error if needed.
+ */
+ protected void cleanupError() throws IOException, InterruptedException {
+ HadoopErrorSimulator.instance().onReduceCleanup();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void setConf(Configuration conf) {
+ wasConfigured = true;
+
+ configError();
+ }
+
+ /**
+ * Simulates configuration error if needed.
+ */
+ protected void configError() {
+ HadoopErrorSimulator.instance().onReduceConfigure();
+ }
+
+ /** {@inheritDoc} */
+ @Override public Configuration getConf() {
+ return null;
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/67b4da70/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/fs/KerberosHadoopFileSystemFactorySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/fs/KerberosHadoopFileSystemFactorySelfTest.java b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/fs/KerberosHadoopFileSystemFactorySelfTest.java
new file mode 100644
index 0000000..8c95a0e
--- /dev/null
+++ b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/fs/KerberosHadoopFileSystemFactorySelfTest.java
@@ -0,0 +1,126 @@
+/*
+ * 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.processors.hadoop.impl.fs;
+
+import java.io.ByteArrayInputStream;
+import java.io.ByteArrayOutputStream;
+import java.io.ObjectInput;
+import java.io.ObjectInputStream;
+import java.io.ObjectOutput;
+import java.io.ObjectOutputStream;
+import java.util.concurrent.Callable;
+
+import org.apache.ignite.hadoop.fs.KerberosHadoopFileSystemFactory;
+import org.apache.ignite.internal.processors.hadoop.delegate.HadoopDelegateUtils;
+import org.apache.ignite.internal.processors.hadoop.delegate.HadoopFileSystemFactoryDelegate;
+import org.apache.ignite.testframework.GridTestUtils;
+import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
+import org.junit.Assert;
+
+/**
+ * Tests KerberosHadoopFileSystemFactory.
+ */
+public class KerberosHadoopFileSystemFactorySelfTest extends GridCommonAbstractTest {
+ /**
+ * Test parameters validation.
+ *
+ * @throws Exception If failed.
+ */
+ public void testParameters() throws Exception {
+ checkParameters(null, null, -1);
+
+ checkParameters(null, null, 100);
+ checkParameters(null, "b", -1);
+ checkParameters("a", null, -1);
+
+ checkParameters(null, "b", 100);
+ checkParameters("a", null, 100);
+ checkParameters("a", "b", -1);
+ }
+
+ /**
+ * Check parameters.
+ *
+ * @param keyTab Key tab.
+ * @param keyTabPrincipal Key tab principal.
+ * @param reloginInterval Re-login interval.
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ private void checkParameters(String keyTab, String keyTabPrincipal, long reloginInterval) {
+ final KerberosHadoopFileSystemFactory fac = new KerberosHadoopFileSystemFactory();
+
+ fac.setKeyTab(keyTab);
+ fac.setKeyTabPrincipal(keyTabPrincipal);
+ fac.setReloginInterval(reloginInterval);
+
+ GridTestUtils.assertThrows(null, new Callable<Object>() {
+ @Override public Object call() throws Exception {
+ HadoopFileSystemFactoryDelegate delegate = HadoopDelegateUtils.fileSystemFactoryDelegate(fac);
+
+ delegate.start();
+
+ return null;
+ }
+ }, IllegalArgumentException.class, null);
+ }
+
+ /**
+ * Checks serializatuion and deserialization of the secure factory.
+ *
+ * @throws Exception If failed.
+ */
+ public void testSerialization() throws Exception {
+ KerberosHadoopFileSystemFactory fac = new KerberosHadoopFileSystemFactory();
+
+ checkSerialization(fac);
+
+ fac = new KerberosHadoopFileSystemFactory();
+
+ fac.setUri("igfs://igfs@localhost:10500/");
+ fac.setConfigPaths("/a/core-sute.xml", "/b/mapred-site.xml");
+ fac.setKeyTabPrincipal("foo");
+ fac.setKeyTab("/etc/krb5.keytab");
+ fac.setReloginInterval(30 * 60 * 1000L);
+
+ checkSerialization(fac);
+ }
+
+ /**
+ * Serializes the factory,
+ *
+ * @param fac The facory to check.
+ * @throws Exception If failed.
+ */
+ private void checkSerialization(KerberosHadoopFileSystemFactory fac) throws Exception {
+ ByteArrayOutputStream baos = new ByteArrayOutputStream();
+
+ ObjectOutput oo = new ObjectOutputStream(baos);
+
+ oo.writeObject(fac);
+
+ ObjectInput in = new ObjectInputStream(new ByteArrayInputStream(baos.toByteArray()));
+
+ KerberosHadoopFileSystemFactory fac2 = (KerberosHadoopFileSystemFactory)in.readObject();
+
+ assertEquals(fac.getUri(), fac2.getUri());
+ Assert.assertArrayEquals(fac.getConfigPaths(), fac2.getConfigPaths());
+ assertEquals(fac.getKeyTab(), fac2.getKeyTab());
+ assertEquals(fac.getKeyTabPrincipal(), fac2.getKeyTabPrincipal());
+ assertEquals(fac.getReloginInterval(), fac2.getReloginInterval());
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/67b4da70/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1DualAbstractTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1DualAbstractTest.java b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1DualAbstractTest.java
new file mode 100644
index 0000000..a585e54
--- /dev/null
+++ b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1DualAbstractTest.java
@@ -0,0 +1,157 @@
+/*
+ * 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.processors.hadoop.impl.igfs;
+
+import java.io.IOException;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.hadoop.fs.CachingHadoopFileSystemFactory;
+import org.apache.ignite.hadoop.fs.IgniteHadoopIgfsSecondaryFileSystem;
+import org.apache.ignite.hadoop.util.ChainedUserNameMapper;
+import org.apache.ignite.hadoop.util.KerberosUserNameMapper;
+import org.apache.ignite.hadoop.util.UserNameMapper;
+import org.apache.ignite.igfs.IgfsIpcEndpointConfiguration;
+import org.apache.ignite.igfs.IgfsIpcEndpointType;
+import org.apache.ignite.igfs.IgfsMode;
+import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystem;
+import org.apache.ignite.internal.processors.igfs.IgfsDualAbstractSelfTest;
+import org.apache.ignite.lifecycle.LifecycleAware;
+import org.jetbrains.annotations.Nullable;
+
+import static org.apache.ignite.IgniteFileSystem.IGFS_SCHEME;
+import static org.apache.ignite.igfs.IgfsMode.PRIMARY;
+
+/**
+ * Abstract test for Hadoop 1.0 file system stack.
+ */
+public abstract class Hadoop1DualAbstractTest extends IgfsDualAbstractSelfTest {
+ /** Secondary grid name */
+ private static final String GRID_NAME = "grid_secondary";
+
+ /** Secondary file system name */
+ private static final String IGFS_NAME = "igfs_secondary";
+
+ /** Secondary file system REST endpoint port */
+ private static final int PORT = 11500;
+
+ /** Secondary file system REST endpoint configuration map. */
+ private static final IgfsIpcEndpointConfiguration SECONDARY_REST_CFG = new IgfsIpcEndpointConfiguration() {{
+ setType(IgfsIpcEndpointType.TCP);
+ setPort(PORT);
+ }};
+
+ /** Secondary file system authority. */
+ private static final String SECONDARY_AUTHORITY = IGFS_NAME + ":" + GRID_NAME + "@127.0.0.1:" + PORT;
+
+ /** Secondary Fs configuration full path. */
+ protected String secondaryConfFullPath;
+
+ /** Secondary Fs URI. */
+ protected String secondaryUri;
+
+ /** Constructor. */
+ public Hadoop1DualAbstractTest(IgfsMode mode) {
+ super(mode);
+ }
+
+ /**
+ * Creates secondary filesystems.
+ * @return IgfsSecondaryFileSystem
+ * @throws Exception On failure.
+ */
+ @Override protected IgfsSecondaryFileSystem createSecondaryFileSystemStack() throws Exception {
+ startUnderlying();
+
+ prepareConfiguration();
+
+ KerberosUserNameMapper mapper1 = new KerberosUserNameMapper();
+
+ mapper1.setRealm("TEST.COM");
+
+ TestUserNameMapper mapper2 = new TestUserNameMapper();
+
+ ChainedUserNameMapper mapper = new ChainedUserNameMapper();
+
+ mapper.setMappers(mapper1, mapper2);
+
+ CachingHadoopFileSystemFactory factory = new CachingHadoopFileSystemFactory();
+
+ factory.setUri(secondaryUri);
+ factory.setConfigPaths(secondaryConfFullPath);
+ factory.setUserNameMapper(mapper);
+
+ IgniteHadoopIgfsSecondaryFileSystem second = new IgniteHadoopIgfsSecondaryFileSystem();
+
+ second.setFileSystemFactory(factory);
+
+ igfsSecondary = new HadoopIgfsSecondaryFileSystemTestAdapter(factory);
+
+ return second;
+ }
+
+ /**
+ * Starts underlying Ignite process.
+ * @throws IOException On failure.
+ */
+ protected void startUnderlying() throws Exception {
+ startGridWithIgfs(GRID_NAME, IGFS_NAME, PRIMARY, null, SECONDARY_REST_CFG, secondaryIpFinder);
+ }
+
+ /**
+ * Prepares Fs configuration.
+ * @throws IOException On failure.
+ */
+ protected void prepareConfiguration() throws IOException {
+ Configuration secondaryConf = HadoopSecondaryFileSystemConfigurationTest.configuration(IGFS_SCHEME, SECONDARY_AUTHORITY, true, true);
+
+ secondaryConf.setInt("fs.igfs.block.size", 1024);
+
+ secondaryConfFullPath = HadoopSecondaryFileSystemConfigurationTest.writeConfiguration(secondaryConf, HadoopSecondaryFileSystemConfigurationTest.SECONDARY_CFG_PATH);
+
+ secondaryUri = HadoopSecondaryFileSystemConfigurationTest.mkUri(IGFS_SCHEME, SECONDARY_AUTHORITY);
+ }
+
+ /**
+ * Test user name mapper.
+ */
+ private static class TestUserNameMapper implements UserNameMapper, LifecycleAware {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** Started flag. */
+ private boolean started;
+
+ /** {@inheritDoc} */
+ @Nullable @Override public String map(String name) {
+ assert started;
+ assert name != null && name.contains("@");
+
+ return name.substring(0, name.indexOf("@"));
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start() throws IgniteException {
+ started = true;
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() throws IgniteException {
+ // No-op.
+ }
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/67b4da70/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualAsyncTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualAsyncTest.java b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualAsyncTest.java
new file mode 100644
index 0000000..97cc7e9
--- /dev/null
+++ b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualAsyncTest.java
@@ -0,0 +1,32 @@
+/*
+ * 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.processors.hadoop.impl.igfs;
+
+import org.apache.ignite.igfs.IgfsMode;
+
+/**
+ * DUAL_ASYNC mode test.
+ */
+public class Hadoop1OverIgfsDualAsyncTest extends Hadoop1DualAbstractTest {
+ /**
+ * Constructor.
+ */
+ public Hadoop1OverIgfsDualAsyncTest() {
+ super(IgfsMode.DUAL_ASYNC);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/67b4da70/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualSyncTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualSyncTest.java b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualSyncTest.java
new file mode 100644
index 0000000..12036bc
--- /dev/null
+++ b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/Hadoop1OverIgfsDualSyncTest.java
@@ -0,0 +1,32 @@
+/*
+ * 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.processors.hadoop.impl.igfs;
+
+import org.apache.ignite.igfs.IgfsMode;
+
+/**
+ * DUAL_SYNC mode.
+ */
+public class Hadoop1OverIgfsDualSyncTest extends Hadoop1DualAbstractTest {
+ /**
+ * Constructor.
+ */
+ public Hadoop1OverIgfsDualSyncTest() {
+ super(IgfsMode.DUAL_SYNC);
+ }
+}
\ No newline at end of file
http://git-wip-us.apache.org/repos/asf/ignite/blob/67b4da70/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/HadoopFIleSystemFactorySelfTest.java
----------------------------------------------------------------------
diff --git a/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/HadoopFIleSystemFactorySelfTest.java b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/HadoopFIleSystemFactorySelfTest.java
new file mode 100644
index 0000000..7cf7f2d
--- /dev/null
+++ b/modules/hadoop-impl/src/test/java/org/apache/ignite/internal/processors/hadoop/impl/igfs/HadoopFIleSystemFactorySelfTest.java
@@ -0,0 +1,345 @@
+/*
+ * 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.processors.hadoop.impl.igfs;
+
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.fs.FileSystem;
+import org.apache.hadoop.fs.Path;
+import org.apache.ignite.IgniteException;
+import org.apache.ignite.cache.CacheWriteSynchronizationMode;
+import org.apache.ignite.configuration.CacheConfiguration;
+import org.apache.ignite.configuration.FileSystemConfiguration;
+import org.apache.ignite.configuration.IgniteConfiguration;
+import org.apache.ignite.hadoop.fs.CachingHadoopFileSystemFactory;
+import org.apache.ignite.hadoop.fs.HadoopFileSystemFactory;
+import org.apache.ignite.hadoop.fs.IgniteHadoopIgfsSecondaryFileSystem;
+import org.apache.ignite.hadoop.fs.v1.IgniteHadoopFileSystem;
+import org.apache.ignite.igfs.IgfsGroupDataBlocksKeyMapper;
+import org.apache.ignite.igfs.IgfsIpcEndpointConfiguration;
+import org.apache.ignite.igfs.IgfsIpcEndpointType;
+import org.apache.ignite.igfs.IgfsMode;
+import org.apache.ignite.igfs.IgfsPath;
+import org.apache.ignite.igfs.secondary.IgfsSecondaryFileSystem;
+import org.apache.ignite.internal.processors.hadoop.delegate.HadoopDelegateUtils;
+import org.apache.ignite.internal.processors.hadoop.delegate.HadoopFileSystemFactoryDelegate;
+import org.apache.ignite.internal.processors.igfs.IgfsCommonAbstractTest;
+import org.apache.ignite.internal.processors.igfs.IgfsEx;
+import org.apache.ignite.internal.util.typedef.G;
+import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.ignite.lifecycle.LifecycleAware;
+import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
+import org.apache.ignite.spi.discovery.tcp.ipfinder.vm.TcpDiscoveryVmIpFinder;
+import org.jetbrains.annotations.Nullable;
+
+import java.io.File;
+import java.io.FileOutputStream;
+import java.io.IOException;
+import java.net.URI;
+import java.util.concurrent.atomic.AtomicInteger;
+
+import static org.apache.ignite.cache.CacheAtomicityMode.TRANSACTIONAL;
+import static org.apache.ignite.cache.CacheMode.PARTITIONED;
+import static org.apache.ignite.cache.CacheMode.REPLICATED;
+
+/**
+ * Tests for Hadoop file system factory.
+ */
+public class HadoopFIleSystemFactorySelfTest extends IgfsCommonAbstractTest {
+ /** Amount of "start" invocations */
+ private static final AtomicInteger START_CNT = new AtomicInteger();
+
+ /** Amount of "stop" invocations */
+ private static final AtomicInteger STOP_CNT = new AtomicInteger();
+
+ /** Path to secondary file system configuration. */
+ private static final String SECONDARY_CFG_PATH = "/work/core-site-HadoopFIleSystemFactorySelfTest.xml";
+
+ /** IGFS path for DUAL mode. */
+ private static final Path PATH_DUAL = new Path("/ignite/sync/test_dir");
+
+ /** IGFS path for PROXY mode. */
+ private static final Path PATH_PROXY = new Path("/ignite/proxy/test_dir");
+
+ /** IGFS path for DUAL mode. */
+ private static final IgfsPath IGFS_PATH_DUAL = new IgfsPath("/ignite/sync/test_dir");
+
+ /** IGFS path for PROXY mode. */
+ private static final IgfsPath IGFS_PATH_PROXY = new IgfsPath("/ignite/proxy/test_dir");
+
+ /** Secondary IGFS. */
+ private IgfsEx secondary;
+
+ /** Primary IGFS. */
+ private IgfsEx primary;
+
+ /** {@inheritDoc} */
+ @Override protected void beforeTest() throws Exception {
+ super.beforeTest();
+
+ START_CNT.set(0);
+ STOP_CNT.set(0);
+
+ secondary = startSecondary();
+ primary = startPrimary();
+ }
+
+ /** {@inheritDoc} */
+ @Override protected void afterTest() throws Exception {
+ super.afterTest();
+
+ secondary = null;
+ primary = null;
+
+ stopAllGrids();
+ }
+
+ /**
+ * Test custom factory.
+ *
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("ThrowableResultOfMethodCallIgnored")
+ public void testCustomFactory() throws Exception {
+ assert START_CNT.get() == 1;
+ assert STOP_CNT.get() == 0;
+
+ // Use IGFS directly.
+ primary.mkdirs(IGFS_PATH_DUAL);
+
+ assert primary.exists(IGFS_PATH_DUAL);
+ assert secondary.exists(IGFS_PATH_DUAL);
+
+ // Create remote instance.
+ FileSystem fs = FileSystem.get(URI.create("igfs://primary:primary@127.0.0.1:10500/"), baseConfiguration());
+
+ // Ensure lifecycle callback was invoked.
+ assert START_CNT.get() == 2;
+ assert STOP_CNT.get() == 0;
+
+ // Check file system operations.
+ assert fs.exists(PATH_DUAL);
+
+ assert fs.delete(PATH_DUAL, true);
+ assert !primary.exists(IGFS_PATH_DUAL);
+ assert !secondary.exists(IGFS_PATH_DUAL);
+ assert !fs.exists(PATH_DUAL);
+
+ assert fs.mkdirs(PATH_DUAL);
+ assert primary.exists(IGFS_PATH_DUAL);
+ assert secondary.exists(IGFS_PATH_DUAL);
+ assert fs.exists(PATH_DUAL);
+
+ assert fs.mkdirs(PATH_PROXY);
+ assert secondary.exists(IGFS_PATH_PROXY);
+ assert fs.exists(PATH_PROXY);
+
+ // Close file system and ensure that associated factory was notified.
+ fs.close();
+
+ assert START_CNT.get() == 2;
+ assert STOP_CNT.get() == 1;
+
+ // Stop primary node and ensure that base factory was notified.
+ G.stop(primary.context().kernalContext().grid().name(), true);
+
+ assert START_CNT.get() == 2;
+ assert STOP_CNT.get() == 2;
+ }
+
+ /**
+ * Start secondary IGFS.
+ *
+ * @return IGFS.
+ * @throws Exception If failed.
+ */
+ private static IgfsEx startSecondary() throws Exception {
+ return start("secondary", 11500, IgfsMode.PRIMARY, null);
+ }
+
+ /**
+ * Start primary IGFS.
+ *
+ * @return IGFS.
+ * @throws Exception If failed.
+ */
+ private static IgfsEx startPrimary() throws Exception {
+ // Prepare configuration.
+ Configuration conf = baseConfiguration();
+
+ conf.set("fs.defaultFS", "igfs://secondary:secondary@127.0.0.1:11500/");
+
+ writeConfigurationToFile(conf);
+
+ // Get file system instance to be used.
+ CachingHadoopFileSystemFactory delegate = new CachingHadoopFileSystemFactory();
+
+ delegate.setUri("igfs://secondary:secondary@127.0.0.1:11500/");
+ delegate.setConfigPaths(SECONDARY_CFG_PATH);
+
+ // Configure factory.
+ TestFactory factory = new TestFactory(delegate);
+
+ // Configure file system.
+ IgniteHadoopIgfsSecondaryFileSystem secondaryFs = new IgniteHadoopIgfsSecondaryFileSystem();
+
+ secondaryFs.setFileSystemFactory(factory);
+
+ // Start.
+ return start("primary", 10500, IgfsMode.DUAL_ASYNC, secondaryFs);
+ }
+
+ /**
+ * Start Ignite node with IGFS instance.
+ *
+ * @param name Node and IGFS name.
+ * @param endpointPort Endpoint port.
+ * @param dfltMode Default path mode.
+ * @param secondaryFs Secondary file system.
+ * @return Igfs instance.
+ */
+ private static IgfsEx start(String name, int endpointPort, IgfsMode dfltMode,
+ @Nullable IgfsSecondaryFileSystem secondaryFs) {
+ IgfsIpcEndpointConfiguration endpointCfg = new IgfsIpcEndpointConfiguration();
+
+ endpointCfg.setType(IgfsIpcEndpointType.TCP);
+ endpointCfg.setHost("127.0.0.1");
+ endpointCfg.setPort(endpointPort);
+
+ FileSystemConfiguration igfsCfg = new FileSystemConfiguration();
+
+ igfsCfg.setDataCacheName("dataCache");
+ igfsCfg.setMetaCacheName("metaCache");
+ igfsCfg.setName(name);
+ igfsCfg.setDefaultMode(dfltMode);
+ igfsCfg.setIpcEndpointConfiguration(endpointCfg);
+ igfsCfg.setSecondaryFileSystem(secondaryFs);
+ igfsCfg.setInitializeDefaultPathModes(true);
+
+ CacheConfiguration dataCacheCfg = defaultCacheConfiguration();
+
+ dataCacheCfg.setName("dataCache");
+ dataCacheCfg.setCacheMode(PARTITIONED);
+ dataCacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ dataCacheCfg.setAffinityMapper(new IgfsGroupDataBlocksKeyMapper(2));
+ dataCacheCfg.setBackups(0);
+ dataCacheCfg.setAtomicityMode(TRANSACTIONAL);
+ dataCacheCfg.setOffHeapMaxMemory(0);
+
+ CacheConfiguration metaCacheCfg = defaultCacheConfiguration();
+
+ metaCacheCfg.setName("metaCache");
+ metaCacheCfg.setCacheMode(REPLICATED);
+ metaCacheCfg.setWriteSynchronizationMode(CacheWriteSynchronizationMode.FULL_SYNC);
+ metaCacheCfg.setAtomicityMode(TRANSACTIONAL);
+
+ IgniteConfiguration cfg = new IgniteConfiguration();
+
+ cfg.setGridName(name);
+
+ TcpDiscoverySpi discoSpi = new TcpDiscoverySpi();
+
+ discoSpi.setIpFinder(new TcpDiscoveryVmIpFinder(true));
+
+ cfg.setDiscoverySpi(discoSpi);
+ cfg.setCacheConfiguration(dataCacheCfg, metaCacheCfg);
+ cfg.setFileSystemConfiguration(igfsCfg);
+
+ cfg.setLocalHost("127.0.0.1");
+ cfg.setConnectorConfiguration(null);
+
+ return (IgfsEx)G.start(cfg).fileSystem(name);
+ }
+
+ /**
+ * Create base FileSystem configuration.
+ *
+ * @return Configuration.
+ */
+ private static Configuration baseConfiguration() {
+ Configuration conf = new Configuration();
+
+ conf.set("fs.igfs.impl", IgniteHadoopFileSystem.class.getName());
+
+ return conf;
+ }
+
+ /**
+ * Write configuration to file.
+ *
+ * @param conf Configuration.
+ * @throws Exception If failed.
+ */
+ @SuppressWarnings("ResultOfMethodCallIgnored")
+ private static void writeConfigurationToFile(Configuration conf) throws Exception {
+ final String path = U.getIgniteHome() + SECONDARY_CFG_PATH;
+
+ File file = new File(path);
+
+ file.delete();
+
+ assertFalse(file.exists());
+
+ try (FileOutputStream fos = new FileOutputStream(file)) {
+ conf.writeXml(fos);
+ }
+
+ assertTrue(file.exists());
+ }
+
+ /**
+ * Test factory.
+ */
+ private static class TestFactory implements HadoopFileSystemFactory, LifecycleAware {
+ /** */
+ private static final long serialVersionUID = 0L;
+
+ /** File system factory. */
+ private CachingHadoopFileSystemFactory factory;
+
+ /** File system. */
+ private transient HadoopFileSystemFactoryDelegate delegate;
+
+ /**
+ * Constructor.
+ *
+ * @param factory File system factory.
+ */
+ public TestFactory(CachingHadoopFileSystemFactory factory) {
+ this.factory = factory;
+ }
+
+ /** {@inheritDoc} */
+ @Override public Object get(String usrName) throws IOException {
+ return delegate.get(usrName);
+ }
+
+ /** {@inheritDoc} */
+ @Override public void start() throws IgniteException {
+ delegate = HadoopDelegateUtils.fileSystemFactoryDelegate(factory);
+
+ delegate.start();
+
+ START_CNT.incrementAndGet();
+ }
+
+ /** {@inheritDoc} */
+ @Override public void stop() throws IgniteException {
+ STOP_CNT.incrementAndGet();
+ }
+ }
+}