IGNITE-13581 Capture Data Change 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.ChangeDataCaptureCommandLineStartup"
+
+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 "$@"

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

@@ -0,0 +1,66 @@
+/*
+ * 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 this} will be returned.
+     * If updated comes from the other cluster using {@link IgniteInternalCache#putAllConflict(Map)} then
+     * @return Replication version.
+     * @see IgniteInternalCache#putAllConflict(Map)
+     * @see IgniteInternalCache#removeAllConflict(Map)
+     */
+    public CacheEntryVersion otherClusterVersion();
+}

modules/core/src/main/java/org/apache/ignite/cdc/ChangeDataCaptureConfiguration.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.ChangeDataCapture;
+import org.apache.ignite.lang.IgniteExperimental;
+
+/**
+ * This class defines {@link ChangeDataCapture} runtime configuration.
+ * Configuration is passed to {@link ChangeDataCapture} constructor.
+ */
+@IgniteExperimental
+public class ChangeDataCaptureConfiguration {
+    /** */
+    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 ChangeDataCaptureConsumer consumer;
+
+    /** Keep binary flag.<br>Default value {@code true}. */
+    private boolean keepBinary = DFLT_KEEP_BINARY;
+
+    /**
+     * {@link ChangeDataCapture} 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#getChangeDataCaptureWalPath()} 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 ChangeDataCaptureConsumer getConsumer() {
+        return consumer;
+    }
+
+    /** @param consumer CDC consumer. */
+    public void setConsumer(ChangeDataCaptureConsumer 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/ChangeDataCaptureConsumer.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.ChangeDataCapture;
+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 ChangeDataCapture} 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> {@code
+ * public class ChangeDataCaptureConsumer implements ChangeDataCaptureConsumer {
+ *     @LoggerReource
+ *     private IgniteLogger log;
+ *
+ *     ...
+ * }
+ * }</pre>
+ *
+ * Note, consumption of the {@link ChangeDataCaptureEvent} 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 ChangeDataCapture
+ * @see ChangeDataCaptureEvent
+ * @see CacheEntryVersion
+ */
+@IgniteExperimental
+public interface ChangeDataCaptureConsumer {
+    /**
+     * 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<ChangeDataCaptureEvent> events);
+
+    /**
+     * Stops the consumer.
+     * This methods can be invoked only after {@link #start()}.
+     */
+    public void stop();
+}

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

@@ -0,0 +1,79 @@
+/*
+ * 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.io.Serializable;
+import org.apache.ignite.cache.CacheEntryVersion;
+import org.apache.ignite.cache.affinity.Affinity;
+import org.apache.ignite.internal.cdc.ChangeDataCapture;
+import org.apache.ignite.lang.IgniteExperimental;
+import org.apache.ignite.spi.systemview.view.CacheView;
+import org.jetbrains.annotations.Nullable;
+
+/**
+ * Event of single entry change.
+ * Instance presents new value of modified entry.
+ *
+ * @see ChangeDataCapture
+ * @see ChangeDataCaptureConsumer
+ */
+@IgniteExperimental
+public interface ChangeDataCaptureEvent extends Serializable {
+    /**
+     * @return Key for the changed entry.
+     */
+    public Object key();
+
+    /**
+     * @return Value for the changed entry or {@code null} in case of entry removal.
+     */
+    @Nullable public Object value();
+
+    /**
+     * @return {@code True} if event fired on primary node for partition containing this entry.
+     * @see <a href="
+     * https://ignite.apache.org/docs/latest/configuring-caches/configuring-backups#configuring-partition-backups">
+     * Configuring partition backups.</a>
+     */
+    public boolean primary();
+
+    /**
+     * Ignite split dataset into smaller chunks to distribute them across the cluster.
+     * {@link ChangeDataCaptureConsumer} implementations can use {@link #partition()} to split changes processing
+     * in the same way as it done for the cache.
+     *
+     * @return Partition number.
+     * @see Affinity#partition(Object)
+     * @see Affinity#partitions()
+     * @see <a href="https://ignite.apache.org/docs/latest/data-modeling/data-partitioning">Data partitioning</a>
+     * @see <a href="https://ignite.apache.org/docs/latest/data-modeling/affinity-collocation">Affinity collocation</a>
+     */
+    public int partition();
+
+    /**
+     * @return Version of the entry.
+     */
+    public CacheEntryVersion version();
+
+    /**
+     * @return Cache ID.
+     * @see org.apache.ignite.internal.util.typedef.internal.CU#cacheId(String)
+     * @see CacheView#cacheId()
+     */
+    public int cacheId();
+}