IGNITE-23304

modules/api/src/main/java/org/apache/ignite/lang/ErrorGroups.java

@@ -389,9 +389,6 @@ public static class Replicator {
 
         /** Stopping replica exception code. */
         public static final int REPLICA_STOPPING_ERR = REPLICATOR_ERR_GROUP.registerErrorCode((short) 8);
-
-        /** Replication safe time reordering. */
-        public static final int REPLICATION_SAFE_TIME_REORDERING_ERR = REPLICATOR_ERR_GROUP.registerErrorCode((short) 9);
     }
 
     /** Storage error group. */

modules/core/src/main/java/org/apache/ignite/internal/hlc/HybridClock.java

@@ -28,6 +28,13 @@ public interface HybridClock {
      */
     long nowLong();
 
+    /**
+     * Creates a timestamp for new event. A timestamp is guarantied to be unique and monotonically grown and follow the causal.
+     *
+     * @return The hybrid timestamp.
+     */
+    long nowLong(HybridTimestamp causal);
+
     /**
      * Gets a current timestamp as long. It is a fast way to get timestamp because it doesn't have to tick the logical part of the clock.
      * This timestamp is not unique, and equal to or less than that value is returned by {@link this#nowLong()}.
@@ -44,7 +51,16 @@ public interface HybridClock {
     HybridTimestamp now();
 
     /**
-     * Gets a current timestamp. It is a fast way to get timestamp because it doesn't have to tick the logical part of the clock.
+     * Creates a timestamp for new event. A timestamp is guarantied to be unique and monotonically grown and follow the causal.
+     *
+     * @param causal The causal timestamp.
+     *
+     * @return The hybrid timestamp.
+     */
+    HybridTimestamp now(HybridTimestamp causal);
+
+    /**
+     * Gets a current timestamp. It is a fast way to get timestamp because it doesn't have to tick.
      * This timestamp is not unique, and equal to or less than that value is returned by {@link this#now()}.
      *
      * @return The hybrid timestamp.

modules/core/src/main/java/org/apache/ignite/internal/hlc/HybridClockImpl.java

@@ -44,15 +44,6 @@ public class HybridClockImpl implements HybridClock {
 
     private final List<ClockUpdateListener> updateListeners = new CopyOnWriteArrayList<>();
 
-    /**
-     * Returns current physical time in milliseconds.
-     *
-     * @return Current time.
-     */
-    protected long physicalTime() {
-        return System.currentTimeMillis();
-    }
-
     @Override
     public final long nowLong() {
         while (true) {
@@ -72,31 +63,40 @@ public final long nowLong() {
     }
 
     @Override
-    public final long currentLong() {
-        long current = currentTime();
+    public final long nowLong(HybridTimestamp causal) {
+        while (true) {
+            long now = max(currentTime(), causal.longValue());
 
-        return max(latestTime, current);
-    }
+            // Read the latest time after accessing UTC time to reduce contention.
+            long oldLatestTime = latestTime;
 
-    private void notifyUpdateListeners(long newTs) {
-        for (ClockUpdateListener listener : updateListeners) {
-            try {
-                listener.onUpdate(newTs);
-            } catch (Throwable e) {
-                log.error("ClockUpdateListener#onUpdate() failed for {} at {}", e, listener, newTs);
+            if (oldLatestTime >= now) {
+                return LATEST_TIME.incrementAndGet(this);
+            }
 
-                if (e instanceof Error) {
-                    throw e;
-                }
+            if (LATEST_TIME.compareAndSet(this, oldLatestTime, now)) {
+                return now;
             }
         }
     }
 
+    @Override
+    public final long currentLong() {
+        long current = currentTime();
+
+        return max(latestTime, current);
+    }
+
     @Override
     public final HybridTimestamp now() {
         return hybridTimestamp(nowLong());
     }
 
+    @Override
+    public HybridTimestamp now(HybridTimestamp causal) {
+        return hybridTimestamp(nowLong(causal));
+    }
+
     @Override
     public final HybridTimestamp current() {
         return hybridTimestamp(currentLong());
@@ -138,10 +138,33 @@ public final HybridTimestamp update(HybridTimestamp requestTime) {
         }
     }
 
+    /**
+     * Returns current physical time in milliseconds.
+     *
+     * @return Current time.
+     */
+    protected long physicalTime() {
+        return System.currentTimeMillis();
+    }
+
     private long currentTime() {
         return physicalTime() << LOGICAL_TIME_BITS_SIZE;
     }
 
+    private void notifyUpdateListeners(long newTs) {
+        for (ClockUpdateListener listener : updateListeners) {
+            try {
+                listener.onUpdate(newTs);
+            } catch (Throwable e) {
+                log.error("ClockUpdateListener#onUpdate() failed for {} at {}", e, listener, newTs);
+
+                if (e instanceof Error) {
+                    throw e;
+                }
+            }
+        }
+    }
+
     @Override
     public void addUpdateListener(ClockUpdateListener listener) {
         updateListeners.add(listener);

modules/core/src/main/java/org/apache/ignite/internal/util/PendingComparableValuesTracker.java

@@ -110,6 +110,35 @@ public void update(T newValue, @Nullable R futureResult) {
         }
     }
 
+    /**
+     * Strict update with reordering check. Always called from the same updater thread.
+     *
+     * @param newValue New value.
+     * @param futureResult A result that will be used to complete a future returned by the
+     *         {@link PendingComparableValuesTracker#waitFor(Comparable)}.
+     */
+    public void updateStrict(T newValue, @Nullable R futureResult) {
+        if (!busyLock.readLock().tryLock()) {
+            throw new TrackerClosedException();
+        }
+
+        try {
+            Map.Entry<T, @Nullable R> current = this.current;
+
+            IgniteBiTuple<T, @Nullable R> newEntry = new IgniteBiTuple<>(newValue, futureResult);
+
+            // Entries from the same batch receive equal safe timestamps.
+            if (comparator.compare(newEntry, current) < 0) {
+                throw new AssertionError("Reordering detected: [old=" + current.getKey() + ", new=" + newEntry.get1() + ']');
+            }
+
+            CURRENT.set(this, newEntry);
+            completeWaitersOnUpdate(newValue, futureResult);
+        } finally {
+            busyLock.readLock().unlock();
+        }
+    }
+
     /**
      * Provides the future that is completed when this tracker's internal value reaches given one. If the internal value is greater or equal
      * then the given one, returns completed future.

modules/core/src/testFixtures/java/org/apache/ignite/internal/TestHybridClock.java

@@ -21,7 +21,7 @@
 import org.apache.ignite.internal.hlc.HybridClockImpl;
 
 /**
- * Test hybrid clock with custom supplier of current time. TODO delete
+ * Test hybrid clock with custom supplier of current time.
  */
 public class TestHybridClock extends HybridClockImpl {
     /** Supplier of current time in milliseconds. */

modules/distribution-zones/build.gradle

@@ -55,6 +55,7 @@ dependencies {
     testImplementation project(':ignite-metastorage')
     testImplementation project(':ignite-system-view-api')
     testImplementation project(':ignite-configuration-system')
+    testImplementation project(':ignite-replicator')
 
     testImplementation(testFixtures(project(':ignite-core')))
     testImplementation(testFixtures(project(':ignite-catalog')))

modules/metastorage/build.gradle

@@ -28,6 +28,7 @@ dependencies {
     implementation project(':ignite-configuration-system')
     implementation project(':ignite-cluster-management')
     implementation project(':ignite-network-api')
+    implementation project(':ignite-network')
     implementation project(':ignite-vault')
     implementation project(':ignite-raft-api')
     implementation project(':ignite-raft')
@@ -38,6 +39,7 @@ dependencies {
     implementation project(':ignite-failure-handler')
     implementation project(':ignite-metrics')
     implementation project(':ignite-system-disaster-recovery-api')
+    implementation project(':ignite-catalog')
     implementation libs.jetbrains.annotations
     implementation libs.fastutil.core
     implementation libs.auto.service.annotations

modules/metastorage/src/main/java/org/apache/ignite/internal/metastorage/command/MetaStorageWriteCommand.java

@@ -28,6 +28,7 @@
 /** Base meta storage write command. */
 public interface MetaStorageWriteCommand extends WriteCommand {
     /** Time on the initiator node. */
+    @Override
     HybridTimestamp initiatorTime();
 
     /**

modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStream.java

@@ -90,6 +90,10 @@ public interface DirectByteBufferStream {
      */
     void writeInt(int val);
 
+    void writeFixedInt(int val);
+
+    void writeFixedLong(long val);
+
     /**
      * Writes {@code Integer}.
      *
@@ -361,6 +365,10 @@ <K, V> void writeMap(Map<K, V> map, MessageCollectionItemType keyType, MessageCo
      */
     int readInt();
 
+    int readFixedInt();
+
+    long readFixedLong();
+
     /**
      * Reads {@code Integer}.
      *

modules/network/src/main/java/org/apache/ignite/internal/network/direct/stream/DirectByteBufferStreamImplV1.java

@@ -313,6 +313,38 @@ public void writeInt(int val) {
         writeVarInt(val + 1);
     }
 
+    /** {@inheritDoc} */
+    @Override
+    public void writeFixedInt(int val) {
+        lastFinished = remainingInternal() >= Integer.BYTES;
+
+        if (lastFinished) {
+            int pos = buf.position();
+
+            GridUnsafe.putInt(heapArr, baseOff + pos, val);
+
+            pos += Integer.BYTES;
+
+            setPosition(pos);
+        }
+    }
+
+    /** {@inheritDoc} */
+    @Override
+    public void writeFixedLong(long val) {
+        lastFinished = remainingInternal() >= Long.BYTES;
+
+        if (lastFinished) {
+            int pos = buf.position();
+
+            GridUnsafe.putLong(heapArr, baseOff + pos, val);
+
+            pos += Long.BYTES;
+
+            setPosition(pos);
+        }
+    }
+
     @Override
     public void writeBoxedInt(@Nullable Integer val) {
         if (val != null) {
@@ -1123,6 +1155,40 @@ public int readInt() {
         return val;
     }
 
+    @Override
+    public int readFixedInt() {
+        lastFinished = remainingInternal() >= Integer.BYTES;
+
+        int val = 0;
+
+        if (lastFinished) {
+            int pos = buf.position();
+
+            val = GridUnsafe.getInt(heapArr, baseOff + pos);
+
+            setPosition(pos + Integer.BYTES);
+        }
+
+        return val;
+    }
+
+    @Override
+    public long readFixedLong() {
+        lastFinished = remainingInternal() >= Long.BYTES;
+
+        long val = 0;
+
+        if (lastFinished) {
+            int pos = buf.position();
+
+            val = GridUnsafe.getLong(heapArr, baseOff + pos);
+
+            setPosition(pos + Long.BYTES);
+        }
+
+        return val;
+    }
+
     @Override
     public @Nullable Integer readBoxedInt() {
         return readBoxedValue(this::readInt);

modules/partition-replicator/src/main/java/org/apache/ignite/internal/partition/replicator/marshaller/PartitionCommandsMarshaller.java

@@ -37,4 +37,6 @@ public interface PartitionCommandsMarshaller extends Marshaller {
      * @return Catalog version. {@value #NO_VERSION_REQUIRED} if version is not required for the given command.
      */
     int readRequiredCatalogVersion(ByteBuffer raw);
+
+    long readSafeTimestamp(ByteBuffer raw);
 }