diff --git a/config/hadoop/default-config.xml b/config/hadoop/default-config.xml
index 65a281e..e45311e 100644
--- a/config/hadoop/default-config.xml
+++ b/config/hadoop/default-config.xml
@@ -123,11 +123,11 @@
-
+
+
+
+
+
diff --git a/examples/config/filesystem/example-igfs.xml b/examples/config/filesystem/example-igfs.xml
index d8ccd34..7334fee 100644
--- a/examples/config/filesystem/example-igfs.xml
+++ b/examples/config/filesystem/example-igfs.xml
@@ -90,9 +90,9 @@
-->
@@ -101,10 +101,9 @@
-->
diff --git a/modules/core/src/main/java/org/apache/ignite/configuration/FileSystemConfiguration.java b/modules/core/src/main/java/org/apache/ignite/configuration/FileSystemConfiguration.java
index f679fc0..5793df1 100644
--- a/modules/core/src/main/java/org/apache/ignite/configuration/FileSystemConfiguration.java
+++ b/modules/core/src/main/java/org/apache/ignite/configuration/FileSystemConfiguration.java
@@ -34,9 +34,6 @@ public class FileSystemConfiguration {
/** Default file system user name. */
public static final String DFLT_USER_NAME = System.getProperty("user.name", "anonymous");
- /** Default IPC port. */
- public static final int DFLT_IPC_PORT = 10500;
-
/** Default fragmentizer throttling block length. */
public static final long DFLT_FRAGMENTIZER_THROTTLING_BLOCK_LENGTH = 16 * 1024 * 1024;
@@ -109,8 +106,8 @@ public class FileSystemConfiguration {
/** Per node parallel operations. */
private int perNodeParallelBatchCnt = DFLT_PER_NODE_PARALLEL_BATCH_CNT;
- /** IPC endpoint properties to publish IGFS over. */
- private Map ipcEndpointCfg;
+ /** IPC endpoint configuration. */
+ private IgfsIpcEndpointConfiguration ipcEndpointCfg;
/** IPC endpoint enabled flag. */
private boolean ipcEndpointEnabled = DFLT_IPC_ENDPOINT_ENABLED;
@@ -401,52 +398,35 @@ public class FileSystemConfiguration {
}
/**
- * Gets map of IPC endpoint configuration properties. There are 2 different
- * types of endpoint supported: {@code shared-memory}, and {@code TCP}.
- *
- * The following configuration properties are supported for {@code shared-memory}
- * endpoint:
- *
- *
{@code type} - value is {@code shmem} to specify {@code shared-memory} approach.
- *
{@code port} - endpoint port.
- *
{@code size} - memory size allocated for single endpoint communication.
- *
- * {@code tokenDirectoryPath} - path, either absolute or relative to {@code IGNITE_HOME} to
- * store shared memory tokens.
- *
- *
- *
- * The following configuration properties are supported for {@code TCP} approach:
- *
- *
{@code type} - value is {@code tcp} to specify {@code TCP} approach.
- *
{@code port} - endpoint bind port.
- *
- * {@code host} - endpoint bind host. If omitted '127.0.0.1' will be used.
- *
- *
+ * Gets IPC endpoint configuration.
*
- * Note that {@code shared-memory} approach is not supported on Windows environments.
- * In case IGFS is failed to bind to particular port, further attempts will be performed every 3 seconds.
+ * Endpoint is needed for communication between IGFS and {@code IgniteHadoopFileSystem} shipped with Ignite
+ * Hadoop Accelerator.
*
- * @return Map of IPC endpoint configuration properties. In case the value is not set, defaults will be used. Default
- * type for Windows is "tcp", for all other platforms - "shmem". Default port is {@link #DFLT_IPC_PORT}.
+ * @return IPC endpoint configuration.
*/
- @Nullable public Map getIpcEndpointConfiguration() {
+ @Nullable public IgfsIpcEndpointConfiguration getIpcEndpointConfiguration() {
return ipcEndpointCfg;
}
/**
- * Sets IPC endpoint configuration to publish IGFS over.
+ * Sets IPC endpoint configuration.
+ *
+ * Endpoint is needed for communication between IGFS and {@code IgniteHadoopFileSystem} shipped with Ignite
+ * Hadoop Accelerator.
*
- * @param ipcEndpointCfg Map of IPC endpoint config properties.
+ * @param ipcEndpointCfg IPC endpoint configuration.
*/
- public void setIpcEndpointConfiguration(@Nullable Map ipcEndpointCfg) {
+ public void setIpcEndpointConfiguration(@Nullable IgfsIpcEndpointConfiguration ipcEndpointCfg) {
this.ipcEndpointCfg = ipcEndpointCfg;
}
/**
* Get IPC endpoint enabled flag. In case it is set to {@code true} endpoint will be created and bound to specific
* port. Otherwise endpoint will not be created. Default value is {@link #DFLT_IPC_ENDPOINT_ENABLED}.
+ *
+ * Endpoint is needed for communication between IGFS and {@code IgniteHadoopFileSystem} shipped with Ignite
+ * Hadoop Accelerator.
*
* @return {@code True} in case endpoint is enabled.
*/
@@ -456,6 +436,9 @@ public class FileSystemConfiguration {
/**
* Set IPC endpoint enabled flag. See {@link #isIpcEndpointEnabled()}.
+ *
+ * Endpoint is needed for communication between IGFS and {@code IgniteHadoopFileSystem} shipped with Ignite
+ * Hadoop Accelerator.
*
* @param ipcEndpointEnabled IPC endpoint enabled flag.
*/
diff --git a/modules/core/src/main/java/org/apache/ignite/igfs/IgfsIpcEndpointConfiguration.java b/modules/core/src/main/java/org/apache/ignite/igfs/IgfsIpcEndpointConfiguration.java
new file mode 100644
index 0000000..6aaf739
--- /dev/null
+++ b/modules/core/src/main/java/org/apache/ignite/igfs/IgfsIpcEndpointConfiguration.java
@@ -0,0 +1,241 @@
+/*
+ * 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.igfs;
+
+import org.apache.ignite.configuration.*;
+import org.apache.ignite.internal.util.typedef.internal.*;
+
+import static org.apache.ignite.igfs.IgfsIpcEndpointType.*;
+
+/**
+ * IGFS IPC endpoint configuration.
+ */
+public class IgfsIpcEndpointConfiguration {
+ /** Default endpoint type; TCP for Windows, SHMEM otherwise. */
+ public static IgfsIpcEndpointType DFLT_TYP = U.isWindows() ? TCP : SHMEM;
+
+ /** Default host. */
+ public static String DFLT_HOST = "127.0.0.1";
+
+ /** Default port. */
+ public static int DFLT_PORT = 10500;
+
+ /** Default shared memory space in bytes. */
+ public static final int DFLT_MEM_SIZE = 256 * 1024;
+
+ /**
+ * Default token directory. Note that this path is relative to {@code IGNITE_HOME/work} folder
+ * if {@code IGNITE_HOME} system or environment variable specified, otherwise it is relative to
+ * {@code work} folder under system {@code java.io.tmpdir} folder.
+ *
+ * @see IgniteConfiguration#getWorkDirectory()
+ */
+ public static final String DFLT_TOKEN_DIR_PATH = "ipc/shmem";
+
+ /** Endpoint type. */
+ private IgfsIpcEndpointType typ = DFLT_TYP;
+
+ /** Host. */
+ private String host = DFLT_HOST;
+
+ /** Port. */
+ private int port = DFLT_PORT;
+
+ /** Space size. */
+ private int memSize = DFLT_MEM_SIZE;
+
+ /** Token directory path. */
+ private String tokenDirPath = DFLT_TOKEN_DIR_PATH;
+
+ /**
+ * Default constructor.
+ */
+ public IgfsIpcEndpointConfiguration() {
+ // No-op.
+ }
+
+ /**
+ * Copying constructor.
+ *
+ * @param cfg Configuration to copy.
+ */
+ public IgfsIpcEndpointConfiguration(IgfsIpcEndpointConfiguration cfg) {
+ typ = cfg.getType();
+ host = cfg.getHost();
+ port = cfg.getPort();
+ memSize = cfg.getMemorySize();
+ tokenDirPath = cfg.getTokenDirectoryPath();
+ }
+
+ /**
+ * Gets endpoint type. There are two endpoints types: {@code SHMEM} working over shared memory, and {@code TCP}
+ * working over sockets.
+ *
+ * Shared memory is recommended approach for Linux-based systems. For Windows TCP is the only available option.
+ *
+ * Defaults to {@link #DFLT_TYP}.
+ *
+ * @return Endpoint type.
+ */
+ public IgfsIpcEndpointType getType() {
+ return typ;
+ }
+
+ /**
+ * Sets endpoint type. There are two endpoints types: {@link IgfsIpcEndpointType#SHMEM} working over shared memory,
+ * and {@link IgfsIpcEndpointType#TCP} working over sockets.
+ *
+ * Shared memory is recommended approach for Linux-based systems. For Windows TCP is the only available option.
+ *
+ * Defaults to {@link #DFLT_TYP}.
+ *
+ * @param typ Endpoint type.
+ */
+ public void setType(IgfsIpcEndpointType typ) {
+ this.typ = typ;
+ }
+
+ /**
+ * Gets the host endpoint is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#TCP} endpoint this is the network interface server socket is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#SHMEM} endpoint socket connection is needed only to perform an initial handshake.
+ * All further communication is performed over shared memory. Therefore, for {@code SHMEM} this value is ignored
+ * and socket will be always bound to {@link #DFLT_HOST}.
+ *
+ * Defaults to {@link #DFLT_HOST}.
+ *
+ * @return Host.
+ */
+ public String getHost() {
+ return host;
+ }
+
+ /**
+ * Sets the host endpoint is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#TCP} endpoint this is the network interface server socket is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#SHMEM} endpoint socket connection is needed only to perform an initial handshake.
+ * All further communication is performed over shared memory. Therefore, for {@code SHMEM} this value is ignored
+ * and socket will be always bound to {@link #DFLT_HOST}.
+ *
+ * Defaults to {@link #DFLT_HOST}.
+ *
+ * @param host Host.
+ */
+ public void setHost(String host) {
+ this.host = host;
+ }
+
+ /**
+ * Gets the port endpoint is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#TCP} endpoint this is the port server socket is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#SHMEM} endpoint socket connection is needed only to perform an initial handshake.
+ * All further communication is performed over shared memory.
+ *
+ * Defaults to {@link #DFLT_PORT}.
+ *
+ * @return Port.
+ */
+ public int getPort() {
+ return port;
+ }
+
+ /**
+ * Sets the port endpoint is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#TCP} endpoint this is the port server socket is bound to.
+ *
+ * For {@link IgfsIpcEndpointType#SHMEM} endpoint socket connection is needed only to perform an initial handshake.
+ * All further communication is performed over shared memory.
+ *
+ * Defaults to {@link #DFLT_PORT}.
+ *
+ * @param port Port.
+ */
+ public void setPort(int port) {
+ this.port = port;
+ }
+
+ /**
+ * Gets shared memory size in bytes allocated for endpoint communication.
+ *
+ * Ignored for {@link IgfsIpcEndpointType#TCP} endpoint.
+ *
+ * Defaults to {@link #DFLT_MEM_SIZE}.
+ *
+ * @return Shared memory size.
+ */
+ public int getMemorySize() {
+ return memSize;
+ }
+
+ /**
+ * Sets shared memory size in bytes allocated for endpoint communication.
+ *
+ * Ignored for {@link IgfsIpcEndpointType#TCP} endpoint.
+ *
+ * Note that this path is relative to {@code IGNITE_HOME/work} folder if {@code IGNITE_HOME} system or environment
+ * variable specified, otherwise it is relative to {@code work} folder under system {@code java.io.tmpdir} folder.
+ *
+ * Ignored for {@link IgfsIpcEndpointType#TCP} endpoint.
+ *
+ * Defaults to {@link #DFLT_TOKEN_DIR_PATH}.
+ *
+ * @return Directory where shared memory tokens are stored.
+ */
+ public String getTokenDirectoryPath() {
+ return tokenDirPath;
+ }
+
+ /**
+ * Sets directory where shared memory tokens are stored.
+ *
+ * Note that this path is relative to {@code IGNITE_HOME/work} folder if {@code IGNITE_HOME} system or environment
+ * variable specified, otherwise it is relative to {@code work} folder under system {@code java.io.tmpdir} folder.
+ *
+ * Ignored for {@link IgfsIpcEndpointType#TCP} endpoint.
+ *
+ * Defaults to {@link #DFLT_TOKEN_DIR_PATH}.
+ *
+ * @param tokenDirPath Directory where shared memory tokens are stored.
+ */
+ public void setTokenDirectoryPath(String tokenDirPath) {
+ this.tokenDirPath = tokenDirPath;
+ }
+
+ /** {@inheritDoc} */
+ @Override public String toString() {
+ return S.toString(IgfsIpcEndpointConfiguration.class, this);
+ }
+}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcServerEndpointDeserializer.java b/modules/core/src/main/java/org/apache/ignite/igfs/IgfsIpcEndpointType.java
similarity index 32%
rename from modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcServerEndpointDeserializer.java
rename to modules/core/src/main/java/org/apache/ignite/igfs/IgfsIpcEndpointType.java
index 07bc28b..475d36f 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/util/ipc/IpcServerEndpointDeserializer.java
+++ b/modules/core/src/main/java/org/apache/ignite/igfs/IgfsIpcEndpointType.java
@@ -15,52 +15,15 @@
* limitations under the License.
*/
-package org.apache.ignite.internal.util.ipc;
-
-import org.apache.ignite.*;
-import org.apache.ignite.internal.util.ipc.loopback.*;
-import org.apache.ignite.internal.util.ipc.shmem.*;
-import org.apache.ignite.internal.util.typedef.internal.*;
-
-import java.util.*;
+package org.apache.ignite.igfs;
/**
- * Grid IpcServerEndpoint configuration deserializer.
+ * IGFS endpoint type.
*/
-public class IpcServerEndpointDeserializer {
- /**
- * Deserializes IPC server endpoint config into concrete
- * instance of {@link IpcServerEndpoint}.
- *
- * @param endpointCfg Map with properties of the IPC server endpoint config.
- * @return Deserialized instance of {@link IpcServerEndpoint}.
- * @throws IgniteCheckedException If any problem with configuration properties setting has happened.
- */
- public static IpcServerEndpoint deserialize(Map endpointCfg) throws IgniteCheckedException {
- A.notNull(endpointCfg, "endpointCfg");
-
- String endpointType = endpointCfg.get("type");
-
- if (endpointType == null)
- throw new IgniteCheckedException("Failed to create server endpoint (type is not specified)");
-
- switch (endpointType) {
- case "shmem": {
- IpcSharedMemoryServerEndpoint endpoint = new IpcSharedMemoryServerEndpoint();
-
- endpoint.setupConfiguration(endpointCfg);
-
- return endpoint;
- }
- case "tcp": {
- IpcServerTcpEndpoint endpoint = new IpcServerTcpEndpoint();
-
- endpoint.setupConfiguration(endpointCfg);
+public enum IgfsIpcEndpointType {
+ /** Shared memory endpoint. */
+ SHMEM,
- return endpoint;
- }
- default:
- throw new IgniteCheckedException("Failed to create server endpoint (type is unknown): " + endpointType);
- }
- }
+ /** TCP endpoint. */
+ TCP;
}
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServer.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServer.java
index 1146812..0cd1a62 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServer.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServer.java
@@ -18,6 +18,7 @@
package org.apache.ignite.internal.processors.igfs;
import org.apache.ignite.*;
+import org.apache.ignite.igfs.*;
import org.apache.ignite.internal.*;
import org.apache.ignite.internal.igfs.common.*;
import org.apache.ignite.internal.util.ipc.*;
@@ -31,7 +32,6 @@ import org.jdk8.backport.*;
import org.jetbrains.annotations.*;
import java.io.*;
-import java.util.*;
import static org.apache.ignite.spi.IgnitePortProtocol.*;
@@ -49,7 +49,7 @@ public class IgfsServer {
private final IgfsMarshaller marsh;
/** Endpoint configuration. */
- private final Map endpointCfg;
+ private final IgfsIpcEndpointConfiguration endpointCfg;
/** Server endpoint. */
private IpcServerEndpoint srvEndpoint;
@@ -72,7 +72,7 @@ public class IgfsServer {
* @param endpointCfg Endpoint configuration to start.
* @param mgmt Management flag - if true, server is intended to be started for Visor.
*/
- public IgfsServer(IgfsContext igfsCtx, Map endpointCfg, boolean mgmt) {
+ public IgfsServer(IgfsContext igfsCtx, IgfsIpcEndpointConfiguration endpointCfg, boolean mgmt) {
assert igfsCtx != null;
assert endpointCfg != null;
@@ -91,7 +91,7 @@ public class IgfsServer {
* @throws IgniteCheckedException If failed.
*/
public void start() throws IgniteCheckedException {
- srvEndpoint = IpcServerEndpointDeserializer.deserialize(endpointCfg);
+ srvEndpoint = createEndpoint(endpointCfg, mgmt);
if (U.isWindows() && srvEndpoint instanceof IpcSharedMemoryServerEndpoint)
throw new IgniteCheckedException(IpcSharedMemoryServerEndpoint.class.getSimpleName() +
@@ -135,6 +135,47 @@ public class IgfsServer {
}
/**
+ * Create server IPC endpoint.
+ *
+ * @param endpointCfg Endpoint configuration.
+ * @param mgmt Management flag.
+ * @return Server endpoint.
+ * @throws IgniteCheckedException If failed.
+ */
+ private static IpcServerEndpoint createEndpoint(IgfsIpcEndpointConfiguration endpointCfg, boolean mgmt)
+ throws IgniteCheckedException {
+ A.notNull(endpointCfg, "endpointCfg");
+
+ IgfsIpcEndpointType typ = endpointCfg.getType();
+
+ if (typ == null)
+ throw new IgniteCheckedException("Failed to create server endpoint (type is not specified)");
+
+ switch (typ) {
+ case SHMEM: {
+ IpcSharedMemoryServerEndpoint endpoint = new IpcSharedMemoryServerEndpoint();
+
+ endpoint.setPort(endpointCfg.getPort());
+ endpoint.setSize(endpointCfg.getMemorySize());
+ endpoint.setTokenDirectoryPath(endpointCfg.getTokenDirectoryPath());
+
+ return endpoint;
+ }
+ case TCP: {
+ IpcServerTcpEndpoint endpoint = new IpcServerTcpEndpoint();
+
+ endpoint.setHost(endpointCfg.getHost());
+ endpoint.setPort(endpointCfg.getPort());
+ endpoint.setManagement(mgmt);
+
+ return endpoint;
+ }
+ default:
+ throw new IgniteCheckedException("Failed to create server endpoint (type is unknown): " + typ);
+ }
+ }
+
+ /**
* Callback that is invoked when kernal is ready.
*/
public void onKernalStart() {
diff --git a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServerManager.java b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServerManager.java
index 643eeff..2cd51f8 100644
--- a/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServerManager.java
+++ b/modules/core/src/main/java/org/apache/ignite/internal/processors/igfs/IgfsServerManager.java
@@ -19,6 +19,7 @@ package org.apache.ignite.internal.processors.igfs;
import org.apache.ignite.*;
import org.apache.ignite.configuration.*;
+import org.apache.ignite.igfs.*;
import org.apache.ignite.internal.util.ipc.*;
import org.apache.ignite.internal.util.typedef.*;
import org.apache.ignite.internal.util.typedef.internal.*;
@@ -30,6 +31,7 @@ import java.util.*;
import java.util.concurrent.*;
import static org.apache.ignite.configuration.FileSystemConfiguration.*;
+import static org.apache.ignite.igfs.IgfsIpcEndpointType.*;
/**
* IGFS server manager.
@@ -50,26 +52,23 @@ public class IgfsServerManager extends IgfsManager {
/** {@inheritDoc} */
@Override protected void start0() throws IgniteCheckedException {
FileSystemConfiguration igfsCfg = igfsCtx.configuration();
- Map cfg = igfsCfg.getIpcEndpointConfiguration();
- if (F.isEmpty(cfg)) {
- // Set default configuration.
- cfg = new HashMap<>();
+ if (igfsCfg.isIpcEndpointEnabled()) {
+ IgfsIpcEndpointConfiguration ipcCfg = igfsCfg.getIpcEndpointConfiguration();
- cfg.put("type", U.isWindows() ? "tcp" : "shmem");
- cfg.put("port", String.valueOf(DFLT_IPC_PORT));
- }
+ if (ipcCfg == null)
+ ipcCfg = new IgfsIpcEndpointConfiguration();
- if (igfsCfg.isIpcEndpointEnabled())
- bind(cfg, /*management*/false);
+ bind(ipcCfg, /*management*/false);
+ }
if (igfsCfg.getManagementPort() >= 0) {
- cfg = new HashMap<>();
+ IgfsIpcEndpointConfiguration mgmtIpcCfg = new IgfsIpcEndpointConfiguration();
- cfg.put("type", "tcp");
- cfg.put("port", String.valueOf(igfsCfg.getManagementPort()));
+ mgmtIpcCfg.setType(TCP);
+ mgmtIpcCfg.setPort(igfsCfg.getManagementPort());
- bind(cfg, /*management*/true);
+ bind(mgmtIpcCfg, /*management*/true);
}
if (bindWorker != null)
@@ -84,7 +83,7 @@ public class IgfsServerManager extends IgfsManager {
* @param mgmt {@code True} if endpoint is management.
* @throws IgniteCheckedException If failed.
*/
- private void bind(final Map endpointCfg, final boolean mgmt) throws IgniteCheckedException {
+ private void bind(final IgfsIpcEndpointConfiguration endpointCfg, final boolean mgmt) throws IgniteCheckedException {
if (srvrs == null)
srvrs = new ConcurrentLinkedQueue<>();
@@ -155,7 +154,7 @@ public class IgfsServerManager extends IgfsManager {
@SuppressWarnings("BusyWait")
private class BindWorker extends GridWorker {
/** Configurations to bind. */
- private Collection, Boolean>> bindCfgs = new LinkedList<>();
+ private Collection> bindCfgs = new LinkedList<>();
/**
* Constructor.
@@ -170,7 +169,7 @@ public class IgfsServerManager extends IgfsManager {
* @param cfg Configuration.
* @param mgmt Management flag.
*/
- public void addConfiguration(Map cfg, boolean mgmt) {
+ public void addConfiguration(IgfsIpcEndpointConfiguration cfg, boolean mgmt) {
bindCfgs.add(F.t(cfg, mgmt));
}
@@ -181,10 +180,10 @@ public class IgfsServerManager extends IgfsManager {
while (!isCancelled()) {
Thread.sleep(REBIND_INTERVAL);
- Iterator, Boolean>> it = bindCfgs.iterator();
+ Iterator> it = bindCfgs.iterator();
while (it.hasNext()) {
- IgniteBiTuple