IGNITE-13581 Change Data Capture implementation

bin/ignite-cdc.sh

@@ -0,0 +1,27 @@
+#!/usr/bin/env bash
+
+#
+# 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.
+#
+
+export MAIN_CLASS="org.apache.ignite.startup.cmdline.CdcCommandLineStartup"
+
+if [ "${IGNITE_HOME:-}" = "" ];
+    then IGNITE_HOME_TMP="$(dirname "$(cd "$(dirname "$0")"; "pwd")")";
+    else IGNITE_HOME_TMP=${IGNITE_HOME};
+fi
+
+${IGNITE_HOME_TMP}/bin/ignite.sh "$@"

config/ignite-log4j2.xml

@@ -31,8 +31,8 @@
         <Routing name="FILE">
             <Routes pattern="$${sys:nodeId}">
                 <Route>
-                    <RollingFile name="Rolling-${sys:nodeId}" fileName="${sys:IGNITE_HOME}/work/log/ignite-${sys:nodeId}.log"
-                                 filePattern="${sys:IGNITE_HOME}/work/log/ignite-${sys:nodeId}-%i-%d{yyyy-MM-dd}.log.gz">
+                    <RollingFile name="Rolling-${sys:nodeId}" fileName="${sys:IGNITE_HOME}/work/log/${sys:appId}-${sys:nodeId}.log"
+                                 filePattern="${sys:IGNITE_HOME}/work/log/${sys:appId}-${sys:nodeId}-%i-%d{yyyy-MM-dd}.log.gz">
                         <PatternLayout pattern="[%d{ISO8601}][%-5p][%t][%c{1}]%notEmpty{[%markerSimpleName]} %m%n"/>
                         <Policies>
                             <TimeBasedTriggeringPolicy interval="6" modulate="true" />

config/java.util.logging.properties

@@ -68,7 +68,7 @@ java.util.logging.ConsoleHandler.level=INFO
 # under `$IGNITE_HOME/work/log/` directory. The placeholder `%{id8}` is a truncated node ID.
 #
 org.apache.ignite.logger.java.JavaLoggerFileHandler.formatter=org.apache.ignite.logger.java.JavaLoggerFormatter
-org.apache.ignite.logger.java.JavaLoggerFileHandler.pattern=ignite-%{id8}.%g.log
+org.apache.ignite.logger.java.JavaLoggerFileHandler.pattern=%{app}-%{id8}.%g.log
 org.apache.ignite.logger.java.JavaLoggerFileHandler.level=INFO
 org.apache.ignite.logger.java.JavaLoggerFileHandler.limit=10485760
 org.apache.ignite.logger.java.JavaLoggerFileHandler.count=10

docs/_docs/code-snippets/xml/log4j2-config.xml

@@ -29,8 +29,8 @@
         <Routing name="FILE">
             <Routes pattern="$${sys:nodeId}">
                 <Route>
-                    <RollingFile name="Rolling-${sys:nodeId}" fileName="${sys:IGNITE_HOME}/work/log/ignite-${sys:nodeId}.log"
-                                 filePattern="${sys:IGNITE_HOME}/work/log/ignite-${sys:nodeId}-%i-%d{yyyy-MM-dd}.log.gz">
+                    <RollingFile name="Rolling-${sys:nodeId}" fileName="${sys:IGNITE_HOME}/work/log/${sys:appId}-${sys:nodeId}.log"
+                                 filePattern="${sys:IGNITE_HOME}/work/log/${sys:appId}-${sys:nodeId}-%i-%d{yyyy-MM-dd}.log.gz">
                         <PatternLayout pattern="[%d{ISO8601}][%-5p][%t][%c{1}]%notEmpty{[%markerSimpleName]} %m%n"/>
                         <Policies>
                             <TimeBasedTriggeringPolicy interval="6" modulate="true" />

modules/compatibility/src/test/java/org/apache/ignite/compatibility/persistence/FoldersReuseCompatibilityTest.java

@@ -32,14 +32,14 @@
 import org.apache.ignite.configuration.PersistentStoreConfiguration;
 import org.apache.ignite.internal.IgniteEx;
 import org.apache.ignite.internal.processors.cache.GridCacheAbstractFullApiSelfTest;
-import org.apache.ignite.internal.processors.cache.persistence.filename.PdsConsistentIdProcessor;
+import org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver;
 import org.apache.ignite.internal.util.typedef.internal.U;
 import org.apache.ignite.lang.IgniteInClosure;
 import org.apache.ignite.spi.discovery.tcp.TcpDiscoverySpi;
 import org.jetbrains.annotations.NotNull;
 import org.junit.Test;
 
-import static org.apache.ignite.internal.processors.cache.persistence.filename.PdsConsistentIdProcessor.parseSubFolderName;
+import static org.apache.ignite.internal.processors.cache.persistence.filename.PdsFolderResolver.parseSubFolderName;
 
 /**
  * Test for new and old style persistent storage folders generation and compatible startup of current ignite version
@@ -198,12 +198,12 @@ private void assertNodeIndexesInFolder(Integer... indexes) throws IgniteCheckedE
      * @throws IgniteCheckedException if failed.
      */
     @NotNull private Set<Integer> getAllNodeIndexesInFolder() throws IgniteCheckedException {
-        final File curFolder = new File(U.defaultWorkDirectory(), PdsConsistentIdProcessor.DB_DEFAULT_FOLDER);
+        final File curFolder = new File(U.defaultWorkDirectory(), PdsFolderResolver.DB_DEFAULT_FOLDER);
         final Set<Integer> indexes = new TreeSet<>();
-        final File[] files = curFolder.listFiles(PdsConsistentIdProcessor.DB_SUBFOLDERS_NEW_STYLE_FILTER);
+        final File[] files = curFolder.listFiles(PdsFolderResolver.DB_SUBFOLDERS_NEW_STYLE_FILTER);
 
         for (File file : files) {
-            final PdsConsistentIdProcessor.FolderCandidate uid
+            final PdsFolderResolver.FolderCandidate uid
                 = parseSubFolderName(file, log);
 
             if (uid != null)
@@ -223,7 +223,7 @@ private void assertPdsDirsDefaultExist(String subDirName) throws IgniteCheckedEx
         assertDirectoryExist(DataStorageConfiguration.DFLT_BINARY_METADATA_PATH, subDirName);
         assertDirectoryExist(PersistentStoreConfiguration.DFLT_WAL_STORE_PATH, subDirName);
         assertDirectoryExist(PersistentStoreConfiguration.DFLT_WAL_ARCHIVE_PATH, subDirName);
-        assertDirectoryExist(PdsConsistentIdProcessor.DB_DEFAULT_FOLDER, subDirName);
+        assertDirectoryExist(PdsFolderResolver.DB_DEFAULT_FOLDER, subDirName);
     }
 
     /**

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerClusterByClassTest.java

@@ -1524,7 +1524,8 @@ private void corruptDataEntry(
                 new GridCacheVersion(),
                 0L,
                 partId,
-                updateCntr
+                updateCntr,
+                DataEntry.EMPTY_FLAGS
             );
 
             GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)ctx.shared().database();

modules/control-utility/src/test/java/org/apache/ignite/util/GridCommandHandlerTest.java

@@ -2638,7 +2638,8 @@ private void corruptDataEntry(
                 new GridCacheVersion(),
                 0L,
                 partId,
-                updateCntr
+                updateCntr,
+                DataEntry.EMPTY_FLAGS
             );
 
             GridCacheDatabaseSharedManager db = (GridCacheDatabaseSharedManager)ctx.shared().database();

modules/core/src/main/java/org/apache/ignite/cache/CacheEntryVersion.java

@@ -0,0 +1,67 @@
+/*
+ * 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.cache;
+
+import java.io.Serializable;
+import java.util.Map;
+import org.apache.ignite.internal.processors.cache.CacheConflictResolutionManager;
+import org.apache.ignite.internal.processors.cache.IgniteInternalCache;
+import org.apache.ignite.internal.processors.cache.version.GridCacheVersionManager;
+import org.apache.ignite.lang.IgniteExperimental;
+
+/**
+ * Entry event order.
+ * Two concurrent updates of the same entry can be ordered based on {@link CacheEntryVersion} comparsion.
+ * Greater value means that event occurs later.
+ *
+ * @see CacheConflictResolutionManager
+ * @see GridCacheVersionManager#dataCenterId(byte)
+ */
+@IgniteExperimental
+public interface CacheEntryVersion extends Comparable<CacheEntryVersion>, Serializable {
+    /**
+     * Order of the update. Value is an incremental counter value. Scope of counter is node.
+     * @return Version order.
+     */
+    public long order();
+
+    /** @return Node order on which this version was assigned. */
+    public int nodeOrder();
+
+    /**
+     * Cluster id is a value to distinguish updates in case user wants to aggregate and sort updates from several
+     * Ignite clusters. {@code clusterId} id can be set for the node using
+     * {@link GridCacheVersionManager#dataCenterId(byte)}.
+     *
+     * @return Cluster id.
+     */
+    public byte clusterId();
+
+    /** @return Topology version plus number of seconds from the start time of the first grid node. */
+    public int topologyVersion();
+
+    /**
+     * If source of the update is "local" cluster then {@code null} will be returned.
+     * If updated comes from the other cluster using {@link IgniteInternalCache#putAllConflict(Map)}
+     * then entry version for other cluster.
+     * @return Replication version.
+     * @see IgniteInternalCache#putAllConflict(Map)
+     * @see IgniteInternalCache#removeAllConflict(Map)
+     */
+    public CacheEntryVersion otherClusterVersion();
+}

modules/core/src/main/java/org/apache/ignite/cdc/CdcConfiguration.java

@@ -0,0 +1,101 @@
+/*
+ * 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.cdc;
+
+import org.apache.ignite.configuration.DataStorageConfiguration;
+import org.apache.ignite.internal.cdc.CdcMain;
+import org.apache.ignite.lang.IgniteExperimental;
+
+/**
+ * This class defines {@link CdcMain} runtime configuration.
+ * Configuration is passed to {@link CdcMain} constructor.
+ */
+@IgniteExperimental
+public class CdcConfiguration {
+    /** */
+    private static final int DFLT_LOCK_TIMEOUT = 1000;
+
+    /** */
+    private static final long DFLT_CHECK_FREQ = 1000L;
+
+    /** */
+    private static final boolean DFLT_KEEP_BINARY = true;
+
+    /** Change Data Capture consumer. */
+    private CdcConsumer consumer;
+
+    /** Keep binary flag.<br>Default value {@code true}. */
+    private boolean keepBinary = DFLT_KEEP_BINARY;
+
+    /**
+     * {@link CdcMain} acquire file lock on startup to ensure exclusive consumption.
+     * This property specifies amount of time to wait for lock acquisition.<br>
+     * Default is {@code 1000 ms}.
+     */
+    private long lockTimeout = DFLT_LOCK_TIMEOUT;
+
+    /**
+     * CDC application periodically scans {@link DataStorageConfiguration#getCdcWalPath()} folder to find new WAL segments.
+     * This frequency specify amount of time application sleeps between subsequent checks when no new files available.
+     * Default is {@code 1000 ms}.
+     */
+    private long checkFreq = DFLT_CHECK_FREQ;
+
+    /** @return CDC consumer. */
+    public CdcConsumer getConsumer() {
+        return consumer;
+    }
+
+    /** @param consumer CDC consumer. */
+    public void setConsumer(CdcConsumer consumer) {
+        this.consumer = consumer;
+    }
+
+    /** @return keep binary value. */
+    public boolean isKeepBinary() {
+        return keepBinary;
+    }
+
+    /** @param keepBinary keep binary value. */
+    public void setKeepBinary(boolean keepBinary) {
+        this.keepBinary = keepBinary;
+    }
+
+    /** @return Amount of time to wait for lock acquisition. */
+    public long getLockTimeout() {
+        return lockTimeout;
+    }
+
+    /** @param lockTimeout Amount of time to wait for lock acquisition. */
+    public void setLockTimeout(long lockTimeout) {
+        this.lockTimeout = lockTimeout;
+    }
+
+    /** @return Amount of time application sleeps between subsequent checks when no new files available. */
+    public long getCheckFrequency() {
+        return checkFreq;
+    }
+
+    /**
+     * @param checkFreq Amount of time application sleeps between subsequent checks when no new
+     *                                           files available.
+     */
+    public void setCheckFrequency(long checkFreq) {
+        this.checkFreq = checkFreq;
+    }
+}

modules/core/src/main/java/org/apache/ignite/cdc/CdcConsumer.java

@@ -0,0 +1,77 @@
+/*
+ * 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.cdc;
+
+import java.util.Iterator;
+import org.apache.ignite.IgniteLogger;
+import org.apache.ignite.cache.CacheEntryVersion;
+import org.apache.ignite.internal.cdc.CdcMain;
+import org.apache.ignite.lang.IgniteExperimental;
+import org.apache.ignite.resources.LoggerResource;
+
+/**
+ * Consumer of WAL data change events.
+ * This consumer will receive data change events during {@link CdcMain} application invocation.
+ * The lifecycle of the consumer is the following:
+ * <ul>
+ *     <li>Start of the consumer {@link #start()}.</li>
+ *     <li>Notification of the consumer by the {@link #onEvents(Iterator)} call.</li>
+ *     <li>Stop of the consumer {@link #stop()}.</li>
+ * </ul>
+ *
+ * In case consumer implementation wants to user {@link IgniteLogger}, please, use, {@link LoggerResource} annotation:
+ * <pre>
+ * public class ChangeDataCaptureConsumer implements ChangeDataCaptureConsumer {
+ *     &#64;LoggerResource
+ *     private IgniteLogger log;
+ *
+ *     ...
+ * }
+ * </pre>
+ *
+ * Note, consumption of the {@link CdcEvent} will be started from the last saved offset.
+ * The offset of consumptions is saved on the disk every time {@link #onEvents(Iterator)} returns {@code true}.
+ *
+ * @see CdcMain
+ * @see CdcEvent
+ * @see CacheEntryVersion
+ */
+@IgniteExperimental
+public interface CdcConsumer {
+    /**
+     * Starts the consumer.
+     */
+    public void start();
+
+    /**
+     * Handles entry changes events.
+     * If this method return {@code true} then current offset will be stored
+     * and ongoing notifications after CDC application fail/restart will be started from it.
+     *
+     * @param events Entry change events.
+     * @return {@code True} if current offset should be saved on the disk
+     * to continue from it in case any failures or restart.
+     */
+    public boolean onEvents(Iterator<CdcEvent> events);
+
+    /**
+     * Stops the consumer.
+     * This methods can be invoked only after {@link #start()}.
+     */
+    public void stop();
+}