Nikolay Izhikov

ID	IEP-59
Author	Nikolay Izhikov
Sponsor
Created	14.10.2020
Status	DRAFT

Motivation

Many use-cases build on observation and processing changed records.

These use-cases include but not limited by

Export data into some warehouse, full-text search system or distributed log system.
Online statistics and analytics
Wait and respond to some specific events or data changes.

For now, such scenarios are hard to implement in Ignite.

The only solution that can help with it, for now, is a Continuous Query.

Disadvantages of the CQ in described scenarios:

CQ requires data to be sent over the network
CQ parts (filter, transformer) live inside server node JVM so issues in it may affect server node stability.
Slow CQ listener leads to increasing of the memory consumption of the server node.
Fails of the CQ listener lead to the loss of the events.

The convenient solution should be:

Independence from the server node process (JVM) - issues and failures of the consumer shouldn't lead to server node instability.
Notification guarantees and failover - i.e. track and save a pointer to the last consumed record. Continue notification from this pointer in case of restart.
Resilience for the consumer - it's not an issue when a consumer temporarily consumes slower than data appear.

Description:

IgniteCDC is a new utility that should be run on the server node host. CDC utility watches by the appearance of the WAL archive segments.

On the segment archiving, utility iterates it using the existing WALIterator and notifications CDCConsumer of each record from the segment.

Design choices:

CDC application works as a separate process.
CDC relies on the existing Ignite mechanism - WAL.
IEP Scope - deliver local data change events to a local consumer.
CDC keeps consumer offset in a special file.
WAL process will start from this offset on restart.
To prevent interference between the WAL archive process and CDC Ignite will create a hard link to the newly created segment in a special folder.
After success processing, CDC will delete this link.
Note, data will be removed from the disk only after CDC and Ignite will remove the link to a segment from both corresponding folders.
To manage minimal event gap new configuration timeout introduced - WalForceArchiveTimeout.
Flag to distinguish DataEntry on primary and backup nodes introduced.

CDCConsumer public API interface:

CDCConsumer.java

/** Consumer of data change events. */
@IgniteExperimental
public interface DataChangeListener<K, V> {
    /**
     * @return Consumer ID.
     */
    String id();

    /**
     * Starts the consumer.
     *
     * @param configuration Ignite configuration.
     */
    void start(IgniteConfiguration configuration, IgniteLogger log);

    /**
     * @return {@code True} if entry key and value should be keeped in binary format.
     */
    boolean keepBinary();

    /**
     * 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 commited.
     */
    boolean onChange(Iterable<EntryEvent<K, V>> events);

    /**
     * Stops the consumer.
     * This methods can be invoked only after {@link #start(IgniteConfiguration, IgniteLogger)}.
     */
    void stop();
}

/**
 * Event for single entry change.
 *
 * @param <K> Key type.
 * @param <V> Value type.
 */
@IgniteExperimental
public interface EntryEvent<K, V> {
    /**
     * @return Key for the changed entry.
     */
    public K key();

    /**
     * @return Value for the changed entry.
     */
    public V 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();

    /**
     * @return Operation type.
     */
    EntryEventType operation();

    /**
     * @return Cache ID.
     * @see org.apache.ignite.internal.util.typedef.internal.CU#cacheId(String)
     */
    long cacheId();

    /**
     * @return Expire time.
     */
    long expireTime();
}

Risks and Assumptions

CDC utility will be started and automatically restarted in the case of failure by the OS or some external tools to provide stable change event processing.
CDC feature may be used for the deployment that has WAL only.
At the start of the CDC first consumed event will be the first event available in the WAL archive.
The lag between the record change and CDC consumer notification will depend on segment archiving timeout and requires additional configuration from the user.
CDC failover depends on the WAL archive segment count. If the CDC application will be down a relatively long time it possible that Ignite deletes certain archive segments,
therefore consumer can't continue to receive changed records and must restart from the existing segments.