IGNITE-22977

modules/client-handler/src/testFixtures/java/org/apache/ignite/client/handler/FakePlacementDriver.java

@@ -100,6 +100,13 @@ public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId repli
         return awaitPrimaryReplica(replicationGroupId, timestamp, 0, TimeUnit.MILLISECONDS);
     }
 
+    @Override
+    public ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
+        TablePartitionId id = (TablePartitionId) replicationGroupId;
+
+        return primaryReplicas.get(id.partitionId());
+    }
+
     @Override
     public CompletableFuture<Void> previousPrimaryExpired(ReplicationGroupId grpId) {
         return nullCompletedFuture();

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

@@ -18,7 +18,6 @@
 package org.apache.ignite.internal.hlc;
 
 import static java.lang.Math.max;
-import static java.time.Clock.systemUTC;
 import static org.apache.ignite.internal.hlc.HybridTimestamp.LOGICAL_TIME_BITS_SIZE;
 import static org.apache.ignite.internal.hlc.HybridTimestamp.hybridTimestamp;
 
@@ -29,6 +28,7 @@
 import org.apache.ignite.internal.logger.IgniteLogger;
 import org.apache.ignite.internal.logger.Loggers;
 import org.apache.ignite.internal.tostring.S;
+import org.apache.ignite.internal.util.FastTimestamps;
 
 /**
  * A Hybrid Logical Clock implementation.
@@ -61,7 +61,7 @@ public HybridClockImpl() {
     }
 
     private static long currentTime() {
-        return systemUTC().instant().toEpochMilli() << LOGICAL_TIME_BITS_SIZE;
+        return FastTimestamps.coarseCurrentTimeMillis() << LOGICAL_TIME_BITS_SIZE;
     }
 
     @Override

modules/core/src/main/java/org/apache/ignite/internal/lang/IgniteSystemProperties.java

@@ -73,6 +73,12 @@ public final class IgniteSystemProperties {
     /** Name of the property controlling whether, when a thread assertion is triggered, it should also be written to the log. */
     public static final String THREAD_ASSERTIONS_LOG_BEFORE_THROWING = "THREAD_ASSERTIONS_LOG_BEFORE_THROWING";
 
+    /** Skip replication in a benchmark. */
+    public static final String SKIP_REPLICATION_IN_BENCHMARK = "SKIP_REPLICATION_IN_BENCHMARK";
+
+    /** Skip storage update in a benchmark. */
+    public static final String SKIP_STORAGE_UPDATE_IN_BENCHMARK = "SKIP_STORAGE_UPDATE_IN_BENCHMARK";
+
     /**
      * Enforces singleton.
      */

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

@@ -23,7 +23,7 @@
 public class FastTimestamps {
     private static volatile long coarseCurrentTimeMillis = System.currentTimeMillis();
 
-    private static final long UPDATE_FREQUENCY_MS = 10;
+    private static final long UPDATE_FREQUENCY_MS = 1;
 
     static {
         startUpdater();
@@ -46,7 +46,7 @@ public void run() {
         };
 
         updater.setDaemon(true);
-        updater.setPriority(10);
+        updater.setPriority(Thread.MAX_PRIORITY);
         updater.start();
     }
 

modules/core/src/test/java/org/apache/ignite/internal/hlc/HybridClockTest.java

@@ -21,15 +21,12 @@
 import static org.mockito.ArgumentMatchers.anyLong;
 import static org.mockito.Mockito.mockStatic;
 import static org.mockito.Mockito.never;
-import static org.mockito.Mockito.spy;
 import static org.mockito.Mockito.verify;
-import static org.mockito.Mockito.when;
 
-import java.time.Clock;
-import java.time.Instant;
 import java.util.concurrent.TimeUnit;
 import java.util.function.Supplier;
 import org.apache.ignite.internal.testframework.BaseIgniteAbstractTest;
+import org.apache.ignite.internal.util.FastTimestamps;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
@@ -46,7 +43,7 @@ class HybridClockTest extends BaseIgniteAbstractTest {
     /**
      * Mock of a system clock.
      */
-    private static MockedStatic<Clock> clockMock;
+    private static MockedStatic<FastTimestamps> clockMock;
 
     @Mock
     private ClockUpdateListener updateListener;
@@ -61,7 +58,7 @@ public void afterEach() {
      */
     @Test
     public void testNow() {
-        clockMock = mockToEpochMilli(100);
+        clockMock = mockCurrentTimestamp(100);
 
         HybridClock clock = new HybridClockImpl();
 
@@ -79,7 +76,7 @@ public void testNow() {
      */
     @Test
     public void testTick() {
-        clockMock = mockToEpochMilli(100);
+        clockMock = mockCurrentTimestamp(100);
 
         HybridClock clock = new HybridClockImpl();
 
@@ -108,7 +105,7 @@ public void testTick() {
     private void assertTimestampEquals(long sysTime, HybridTimestamp expTs, Supplier<HybridTimestamp> clo) {
         closeClockMock();
 
-        clockMock = mockToEpochMilli(sysTime);
+        clockMock = mockCurrentTimestamp(sysTime);
 
         assertEquals(expTs, clo.get());
     }
@@ -167,12 +164,10 @@ void updateListenerIsNotUpdatedAfterRemoval() {
         verify(updateListener, never()).onUpdate(anyLong());
     }
 
-    private static MockedStatic<Clock> mockToEpochMilli(long expected) {
-        Clock spyClock = spy(Clock.class);
-        MockedStatic<Clock> clockMock = mockStatic(Clock.class);
+    private static MockedStatic<FastTimestamps> mockCurrentTimestamp(long expected) {
+        MockedStatic<FastTimestamps> clockMock = mockStatic(FastTimestamps.class);
 
-        clockMock.when(Clock::systemUTC).thenReturn(spyClock);
-        when(spyClock.instant()).thenReturn(Instant.ofEpochMilli(expected));
+        clockMock.when(FastTimestamps::coarseCurrentTimeMillis).thenReturn(expected);
 
         return clockMock;
     }

modules/index/src/main/java/org/apache/ignite/internal/index/IndexNodeFinishedRwTransactionsChecker.java

@@ -172,7 +172,7 @@ private void onReceiveIndexNetworkMessage(NetworkMessage message, ClusterNode se
     }
 
     /**
-     * Returns {@code true} iff the requested catalog version is active and all RW transactions started on versions strictly before that
+     * Returns {@code true} if the requested catalog version is active and all RW transactions started on versions strictly before that
      * version have finished on the node.
      *
      * @param catalogVersion Catalog version of interest.

modules/index/src/test/java/org/apache/ignite/internal/index/TestPlacementDriver.java

@@ -32,6 +32,7 @@
 import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
 import org.apache.ignite.internal.replicator.ReplicationGroupId;
 import org.apache.ignite.internal.replicator.TablePartitionId;
+import org.jetbrains.annotations.Nullable;
 
 /** Implementation for tests. */
 class TestPlacementDriver extends AbstractEventProducer<PrimaryReplicaEvent, PrimaryReplicaEventParameters> implements PlacementDriver {
@@ -52,6 +53,11 @@ public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId repli
         return primaryReplicaMetaFutureById.get(replicationGroupId);
     }
 
+    @Override
+    public @Nullable ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
+        return primaryReplicaMetaFutureById.get(replicationGroupId).join();
+    }
+
     @Override
     public CompletableFuture<Void> previousPrimaryExpired(ReplicationGroupId grpId) {
         throw new UnsupportedOperationException();

modules/page-memory/src/test/java/org/apache/ignite/internal/pagememory/persistence/checkpoint/CheckpointMetricsTrackerTest.java

@@ -50,7 +50,7 @@ void testStartAndEndCheckpoint() throws Exception {
 
         assertThat(tracker.checkpointStartTime(), equalTo(checkpointStartTime));
 
-        assertThat(tracker.totalDuration(), greaterThanOrEqualTo(10L));
+        assertThat(tracker.totalDuration(), greaterThanOrEqualTo(1L));
     }
 
     @Test
@@ -97,7 +97,7 @@ void testSplitAndSortCheckpointPages() throws Exception {
 
         tracker.onSplitAndSortCheckpointPagesEnd();
 
-        assertThat(tracker.splitAndSortCheckpointPagesDuration(), greaterThanOrEqualTo(10L));
+        assertThat(tracker.splitAndSortCheckpointPagesDuration(), greaterThanOrEqualTo(1L));
     }
 
     @Test
@@ -114,7 +114,7 @@ void testFsync() throws Exception {
 
         tracker.onCheckpointEnd();
 
-        assertThat(tracker.fsyncDuration(), greaterThanOrEqualTo(10L));
+        assertThat(tracker.fsyncDuration(), greaterThanOrEqualTo(1L));
     }
 
     @Test
@@ -131,7 +131,7 @@ void testPagesWrite() throws Exception {
 
         tracker.onFsyncStart();
 
-        assertThat(tracker.pagesWriteDuration(), greaterThanOrEqualTo(10L));
+        assertThat(tracker.pagesWriteDuration(), greaterThanOrEqualTo(1L));
     }
 
     @Test
@@ -148,7 +148,7 @@ void testOnMarkCheckpointBegin() throws Exception {
 
         tracker.onMarkCheckpointBeginEnd();
 
-        assertThat(tracker.onMarkCheckpointBeginDuration(), greaterThanOrEqualTo(10L));
+        assertThat(tracker.onMarkCheckpointBeginDuration(), greaterThanOrEqualTo(1L));
     }
 
     @Test
@@ -165,7 +165,7 @@ void testWriteLock() throws Exception {
 
         long beforeWriteLockDuration = tracker.beforeWriteLockDuration();
 
-        assertThat(beforeWriteLockDuration, greaterThanOrEqualTo(10L));
+        assertThat(beforeWriteLockDuration, greaterThanOrEqualTo(1L));
         assertThat(tracker.writeLockWaitDuration(), lessThan(0L));
         assertThat(tracker.writeLockHoldDuration(), equalTo(0L));
 
@@ -176,7 +176,7 @@ void testWriteLock() throws Exception {
         long writeLockWaitDuration = tracker.writeLockWaitDuration();
 
         assertThat(tracker.beforeWriteLockDuration(), equalTo(beforeWriteLockDuration));
-        assertThat(writeLockWaitDuration, greaterThanOrEqualTo(10L));
+        assertThat(writeLockWaitDuration, greaterThanOrEqualTo(1L));
         assertThat(tracker.writeLockHoldDuration(), lessThan(0L));
 
         waitForChangeCoarseCurrentTimeMillis();
@@ -185,7 +185,7 @@ void testWriteLock() throws Exception {
 
         assertThat(tracker.beforeWriteLockDuration(), equalTo(beforeWriteLockDuration));
         assertThat(tracker.writeLockWaitDuration(), equalTo(writeLockWaitDuration));
-        assertThat(tracker.writeLockHoldDuration(), greaterThanOrEqualTo(10L));
+        assertThat(tracker.writeLockHoldDuration(), greaterThanOrEqualTo(1L));
     }
 
     private void waitForChangeCoarseCurrentTimeMillis() throws Exception {

modules/partition-replicator/src/integrationTest/java/org/apache/ignite/internal/partition/replicator/utils/TestPlacementDriver.java

@@ -61,6 +61,12 @@ public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId repli
         return getPrimaryReplicaMeta();
     }
 
+    @Override
+    public ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId,
+            HybridTimestamp timestamp) {
+        return getPrimaryReplicaMeta().join();
+    }
+
     @Override
     public CompletableFuture<Void> previousPrimaryExpired(ReplicationGroupId grpId) {
         return nullCompletedFuture();

modules/placement-driver-api/src/main/java/org/apache/ignite/internal/placementdriver/LeasePlacementDriver.java

@@ -24,6 +24,7 @@
 import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEvent;
 import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
 import org.apache.ignite.internal.replicator.ReplicationGroupId;
+import org.jetbrains.annotations.Nullable;
 
 /**
  * Service that provides an ability to await and retrieve primary replicas for replication groups.
@@ -70,6 +71,15 @@ CompletableFuture<ReplicaMeta> awaitPrimaryReplica(
      */
     CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp);
 
+    /**
+     * Returns the current primary replica without waiting.
+     *
+     * @param replicationGroupId Replication group id.
+     * @param timestamp The timestamp.
+     * @return Metadata information or null if not available.
+     */
+    @Nullable ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp);
+
     /**
      * Returns a future that completes when all expiration event {@link PrimaryReplicaEvent#PRIMARY_REPLICA_EXPIRED} listeners of previous
      * primary complete.

modules/placement-driver-api/src/testFixtures/java/org/apache/ignite/internal/placementdriver/TestPlacementDriver.java

@@ -31,6 +31,7 @@
 import org.apache.ignite.internal.placementdriver.event.PrimaryReplicaEventParameters;
 import org.apache.ignite.internal.replicator.ReplicationGroupId;
 import org.apache.ignite.network.ClusterNode;
+import org.jetbrains.annotations.Nullable;
 import org.jetbrains.annotations.TestOnly;
 
 /**
@@ -74,6 +75,11 @@ public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId repli
         return getReplicaMetaFuture();
     }
 
+    @Override
+    public @Nullable ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
+        return getReplicaMetaFuture().join();
+    }
+
     @Override
     public CompletableFuture<Void> previousPrimaryExpired(ReplicationGroupId grpId) {
         return nullCompletedFuture();

modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/PlacementDriverManager.java

@@ -298,6 +298,11 @@ public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId repli
                 return leaseTracker.getPrimaryReplica(replicationGroupId, timestamp);
             }
 
+            @Override
+            public ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
+                return leaseTracker.getCurrentPrimaryReplica(replicationGroupId, timestamp);
+            }
+
             @Override
             public CompletableFuture<Void> previousPrimaryExpired(ReplicationGroupId replicationGroupId) {
                 return leaseTracker.previousPrimaryExpired(replicationGroupId);

modules/placement-driver/src/main/java/org/apache/ignite/internal/placementdriver/leases/LeaseTracker.java

@@ -300,6 +300,17 @@ public CompletableFuture<ReplicaMeta> getPrimaryReplica(ReplicationGroupId repli
         });
     }
 
+    @Override
+    public @Nullable ReplicaMeta getCurrentPrimaryReplica(ReplicationGroupId replicationGroupId, HybridTimestamp timestamp) {
+        Lease lease = getLease(replicationGroupId);
+
+        if (lease.isAccepted() && clockService.after(lease.getExpirationTime(), timestamp)) {
+            return lease;
+        }
+
+        return null;
+    }
+
     /**
      * Helper method that checks whether tracker for given groupId is present in {@code primaryReplicaWaiters} map, whether it's empty
      * and removes it if it's true.