Optimize heartbeat clear. Deprecate use-sync-indexing API

Author: jjezra

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingBase.java

@@ -150,7 +150,7 @@ protected CompletableFuture<FDBStoredRecord<Message>> recordIfInIndexedTypes(FDB
 
     // buildIndexAsync - the main indexing function. Builds and commits indexes asynchronously; throttling to avoid overloading the system.
     @SuppressWarnings("PMD.CloseResource")
-    public CompletableFuture<Void> buildIndexAsync(boolean markReadable, boolean useSyncLock) {
+    public CompletableFuture<Void> buildIndexAsync(boolean markReadable) {
         KeyValueLogMessage message = KeyValueLogMessage.build("build index online",
                 LogMessageKeys.SHOULD_MARK_READABLE, markReadable);
         long startNanos = System.nanoTime();
@@ -179,6 +179,8 @@ public CompletableFuture<Void> buildIndexAsync(boolean markReadable, boolean use
                     }
                     return ret;
                 })
+                // Here: if the heartbeat was *not* cleared while marking the index readable, it would be cleared in
+                // these dedicated transaction. Heartbeat clearing is not a blocker but a "best effort" operation.
                 .thenCompose(ignore -> clearHeartbeats())
                 .handle((ignore, exIgnore) -> {
                     Throwable ex = indexingException.get();
@@ -322,6 +324,7 @@ public CompletableFuture<Boolean> markIndexReadable(boolean markReadablePlease)
             if (ex != null) {
                 throw ex;
             }
+            heartbeat = null; // Here: heartbeats had been successfully cleared. No need to clear again
             return anythingChanged.get();
         });
     }
@@ -332,12 +335,14 @@ private CompletableFuture<Boolean> markIndexReadableSingleTarget(Index index, At
         return getRunner().runAsync(context ->
                 common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
                         .thenCompose(store -> {
+                            clearHeartbeatSingleTarget(store, index);
                             return policy.shouldAllowUniquePendingState(store) ?
                                    store.markIndexReadableOrUniquePending(index) :
                                    store.markIndexReadable(index);
                         })
         ).handle((changed, ex) -> {
             if (ex == null) {
+                heartbeat = null; // Here: all heartbeats were successfully cleared withing the set readable transactions. No need to clear again.
                 if (Boolean.TRUE.equals(changed)) {
                     anythingChanged.set(true);
                 }
@@ -358,7 +363,7 @@ public void enforceStampOverwrite() {
     private CompletableFuture<Void> setIndexingTypeOrThrow(FDBRecordStore store, boolean continuedBuild) {
         // continuedBuild is set if this session isn't a continuation of a previous indexing
         IndexBuildProto.IndexBuildIndexingStamp indexingTypeStamp = getIndexingTypeStamp(store);
-        heartbeat = new IndexingHeartbeat(common.getUuid(), indexingTypeStamp.getMethod());
+        heartbeat = new IndexingHeartbeat(common.getUuid(), indexingTypeStamp.getMethod(), common.config.getLeaseLengthMillis());
 
         return forEachTargetIndex(index -> setIndexingTypeOrThrow(store, continuedBuild, index, indexingTypeStamp));
     }
@@ -850,23 +855,28 @@ private CompletableFuture<Void> updateHeartbeat(boolean validate, FDBRecordStore
     }
 
     private CompletableFuture<Void> clearHeartbeats() {
-        // Here: if the heartbeat was *not* cleared while marking the index readable, it would be cleared in
-        // these dedicated transaction. Heartbeat clearing is not a blocker but a "best effort" operation.
         if (heartbeat == null) {
             return AsyncUtil.DONE;
         }
-        return forEachTargetIndex(this::clearHeartbeatSingleTarget);
+        return forEachTargetIndex(this::clearHeartbeatSingleTarget)
+                .thenAccept(ignore -> heartbeat = null);
     }
 
     private CompletableFuture<Void> clearHeartbeatSingleTarget(Index index) {
         return getRunner().runAsync(context ->
                 common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
                         .thenApply(store -> {
-                            heartbeat.clearHeartbeat(store, index);
+                            clearHeartbeatSingleTarget(store, index);
                             return null;
                         }));
     }
 
+    private void clearHeartbeatSingleTarget(FDBRecordStore store, Index index) {
+        if (heartbeat != null) {
+            heartbeat.clearHeartbeat(store, index);
+        }
+    }
+
 
     private boolean shouldValidate() {
         final long minimalInterval = policy.getCheckIndexingMethodFrequencyMilliseconds();

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/IndexingHeartbeat.java

@@ -33,17 +33,18 @@
 import javax.annotation.Nonnull;
 import java.util.UUID;
 import java.util.concurrent.CompletableFuture;
-import java.util.concurrent.TimeUnit;
 
 public class IndexingHeartbeat {
     // [prefix, xid] -> [indexing-type, genesis time, heartbeat time]
     final UUID sessionId;
     final IndexBuildProto.IndexBuildIndexingStamp.Method indexingMethod;
     final long genesisTimeMilliseconds;
+    final long leaseLength;
 
-    public IndexingHeartbeat(final UUID sessionId, IndexBuildProto.IndexBuildIndexingStamp.Method indexingMethod) {
+    public IndexingHeartbeat(final UUID sessionId, IndexBuildProto.IndexBuildIndexingStamp.Method indexingMethod, long leaseLength) {
         this.sessionId = sessionId;
         this.indexingMethod = indexingMethod;
+        this.leaseLength = leaseLength;
         this.genesisTimeMilliseconds = nowMilliseconds();
     }
 
@@ -94,7 +95,7 @@ private void validateNonCompetingHeartbeat(KeyValue kv) {
             try {
                 final IndexBuildProto.IndexingHeartbeat otherHeartbeat = IndexBuildProto.IndexingHeartbeat.parseFrom(kv.getValue());
                 final long age = now - otherHeartbeat.getHeartbeatTimeMilliseconds();
-                if (age > 0 && age < TimeUnit.SECONDS.toMillis(10)) {
+                if (age > 0 && age < leaseLength) {
                     throw new SynchronizedSessionLockedException("Failed to initialize the session because of an existing session in progress");
                     // TODO: log details
                 }

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexOperationBaseBuilder.java

@@ -20,6 +20,7 @@
 
 package com.apple.foundationdb.record.provider.foundationdb;
 
+import com.apple.foundationdb.annotation.API;
 import com.apple.foundationdb.record.RecordCoreException;
 import com.apple.foundationdb.record.RecordMetaData;
 import com.apple.foundationdb.record.RecordMetaDataProvider;
@@ -715,15 +716,17 @@ public B setTransactionTimeLimitMilliseconds(long timeLimitMilliseconds) {
      * @see SynchronizedSessionRunner
      * @param useSynchronizedSession use synchronize session if true, otherwise false
      * @return this builder
+     *
+     * @deprecated Synchronized sessions are now determined by the indexing method.
      */
+    @API(API.Status.DEPRECATED)
+    @Deprecated(since = "4.4.3.0", forRemoval = true)
     public B setUseSynchronizedSession(boolean useSynchronizedSession) {
-        configBuilder.setUseSynchronizedSession(useSynchronizedSession);
         return self();
     }
 
     /**
      * Set the lease length in milliseconds if the synchronized session is used. The default value is {@link OnlineIndexOperationConfig#DEFAULT_LEASE_LENGTH_MILLIS}.
-     * @see #setUseSynchronizedSession(boolean)
      * @see com.apple.foundationdb.synchronizedsession.SynchronizedSession
      * @param leaseLengthMillis length between last access and lease's end time in milliseconds
      * @return this builder

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexOperationConfig.java

@@ -77,14 +77,13 @@ public class OnlineIndexOperationConfig {
     private final int increaseLimitAfter;
     private final long timeLimitMilliseconds;
     private final long transactionTimeLimitMilliseconds;
-    private final boolean useSynchronizedSession;
     private final long leaseLengthMillis;
 
     public static final long UNLIMITED_TIME = 0;
 
     OnlineIndexOperationConfig(int maxLimit, int initialLimit, int maxRetries, int recordsPerSecond, long progressLogIntervalMillis, int increaseLimitAfter,
                                int maxWriteLimitBytes, long timeLimitMilliseconds, long transactionTimeLimitMilliseconds,
-                               boolean useSynchronizedSession, long leaseLengthMillis) {
+                               long leaseLengthMillis) {
         this.maxLimit = maxLimit;
         this.initialLimit = initialLimit;
         this.maxRetries = maxRetries;
@@ -94,7 +93,6 @@ public class OnlineIndexOperationConfig {
         this.maxWriteLimitBytes = maxWriteLimitBytes;
         this.timeLimitMilliseconds = timeLimitMilliseconds;
         this.transactionTimeLimitMilliseconds = transactionTimeLimitMilliseconds;
-        this.useSynchronizedSession = useSynchronizedSession;
         this.leaseLengthMillis = leaseLengthMillis;
     }
 
@@ -188,10 +186,6 @@ public static Builder newBuilder() {
         return new Builder();
     }
 
-    public boolean shouldUseSynchronizedSession() {
-        return useSynchronizedSession;
-    }
-
     public long getLeaseLengthMillis() {
         return leaseLengthMillis;
     }
@@ -213,7 +207,6 @@ public Builder toBuilder() {
                 .setMaxRetries(this.maxRetries)
                 .setTimeLimitMilliseconds(timeLimitMilliseconds)
                 .setTransactionTimeLimitMilliseconds(this.transactionTimeLimitMilliseconds)
-                .setUseSynchronizedSession(useSynchronizedSession)
                 .setLeaseLengthMillis(leaseLengthMillis);
     }
 
@@ -234,7 +227,6 @@ public static class Builder {
         private long timeLimitMilliseconds = UNLIMITED_TIME;
         private long transactionTimeLimitMilliseconds = DEFAULT_TRANSACTION_TIME_LIMIT;
         private long leaseLengthMillis = DEFAULT_LEASE_LENGTH_MILLIS;
-        private boolean useSynchronizedSession = true;
 
         protected Builder() {
 
@@ -492,15 +484,18 @@ public Builder setTransactionTimeLimitMilliseconds(long timeLimitMilliseconds) {
          * @see SynchronizedSessionRunner
          * @param useSynchronizedSession use synchronize session if true, otherwise false
          * @return this builder
+         *
+         * @deprecated Synchronized sessions are now determined by the indexing method.
          */
+        @API(API.Status.DEPRECATED)
+        @Deprecated(since = "4.4.3.0", forRemoval = true)
         public Builder setUseSynchronizedSession(boolean useSynchronizedSession) {
-            this.useSynchronizedSession = useSynchronizedSession;
+            // no-op
             return this;
         }
 
         /**
          * Set the lease length in milliseconds if the synchronized session is used. By default this is {@link #DEFAULT_LEASE_LENGTH_MILLIS}.
-         * @see #setUseSynchronizedSession(boolean)
          * @see com.apple.foundationdb.synchronizedsession.SynchronizedSession
          * @param leaseLengthMillis length between last access and lease's end time in milliseconds
          * @return this builder
@@ -519,7 +514,7 @@ public Builder setLeaseLengthMillis(long leaseLengthMillis) {
         public OnlineIndexOperationConfig build() {
             return new OnlineIndexOperationConfig(maxLimit, initialLimit, maxRetries, recordsPerSecond, progressLogIntervalMillis, increaseLimitAfter,
                     maxWriteLimitBytes, timeLimitMilliseconds, transactionTimeLimitMilliseconds,
-                    useSynchronizedSession, leaseLengthMillis);
+                    leaseLengthMillis);
         }
     }
 }

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexScrubber.java

@@ -75,7 +75,7 @@ private IndexingBase getScrubber(IndexScrubbingTools.ScrubbingType type, AtomicL
     @Nonnull
     private CompletableFuture<Void> scrubIndexAsync(IndexScrubbingTools.ScrubbingType type, AtomicLong count) {
         return AsyncUtil.composeHandle(
-                getScrubber(type, count).buildIndexAsync(false, common.config.shouldUseSynchronizedSession()),
+                getScrubber(type, count).buildIndexAsync(false),
                 (ignore, ex) -> {
                     if (ex != null) {
                         throw FDBExceptions.wrapException(ex);

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexer.java

@@ -501,11 +501,6 @@ public static CompletableFuture<Boolean> checkAnyOngoingOnlineIndexBuildsAsync(@
      * any retriable errors that it encounters while it runs the build. At the end, it marks the index readable in the
      * store.
      * </p>
-     * <p>
-     * One may consider to set the index state precondition to {@link IndexStatePrecondition#ERROR_IF_DISABLED_CONTINUE_IF_WRITE_ONLY}
-     * and {@link OnlineIndexer.Builder#setUseSynchronizedSession(boolean)} to {@code false}, which makes the indexer
-     * follow the same behavior as before version 2.8.90.0. But it is not recommended.
-     * </p>
      * @return a future that will be ready when the build has completed
      * @throws com.apple.foundationdb.synchronizedsession.SynchronizedSessionLockedException the build is stopped
      * because there may be another build running actively on this index.
@@ -518,8 +513,7 @@ public CompletableFuture<Void> buildIndexAsync() {
     @VisibleForTesting
     @Nonnull
     CompletableFuture<Void> buildIndexAsync(boolean markReadable) {
-        boolean useSyncLock = (!indexingPolicy.isMutual() || fallbackToRecordsScan) && common.config.shouldUseSynchronizedSession();
-        return indexingLauncher(() -> getIndexer().buildIndexAsync(markReadable, useSyncLock));
+        return indexingLauncher(() -> getIndexer().buildIndexAsync(markReadable));
     }
 
     /**
@@ -763,11 +757,6 @@ public Builder setRecordTypes(@Nullable Collection<RecordType> recordTypes) {
          * Set how should {@link #buildIndexAsync()} (or its variations) build the index based on its state. Normally
          * this should be {@link IndexStatePrecondition#BUILD_IF_DISABLED_CONTINUE_BUILD_IF_WRITE_ONLY} if the index is
          * not corrupted.
-         * <p>
-         * One may consider setting it to {@link IndexStatePrecondition#ERROR_IF_DISABLED_CONTINUE_IF_WRITE_ONLY} and
-         * {@link #setUseSynchronizedSession(boolean)} to {@code false}, which makes the indexer follow the same behavior
-         * as before version 2.8.90.0. But it is not recommended.
-         * </p>
          * @see IndexStatePrecondition
          * @param indexStatePrecondition build option to use
          * @return this builder
@@ -1449,7 +1438,6 @@ public Builder setMutualIndexing() {
              * by other threads/processes/systems with the exact same parameters, are attempting to concurrently build this
              * index. To allow that, the indexer will:
              * <ol>
-             *   <li> Avoid the indexing lock - i.e. assume that {@link OnlineIndexer.Builder#setUseSynchronizedSession(boolean)} was called with false</li>
              *   <li> Divide the records space to fragments, then iterate the fragments in a way that minimize the interference, while
              *      indexing each fragment independently.</li>
              *   <li> Handle indexing conflicts, when occurred.</li>

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerBuildIndexTest.java

@@ -180,7 +180,6 @@ <M extends Message> void singleRebuild(
         if (!safeBuild) {
             indexingPolicy.setIfDisabled(OnlineIndexer.IndexingPolicy.DesiredAction.ERROR)
                     .setIfMismatchPrevious(OnlineIndexer.IndexingPolicy.DesiredAction.ERROR);
-            builder.setUseSynchronizedSession(false);
         }
         if (sourceIndex != null) {
             indexingPolicy.setSourceIndex(sourceIndex.getName())

fdb-record-layer-core/src/test/java/com/apple/foundationdb/record/provider/foundationdb/OnlineIndexerIndexFromIndexTest.java

@@ -23,8 +23,6 @@
 import com.apple.foundationdb.record.IndexBuildProto;
 import com.apple.foundationdb.record.RecordCoreException;
 import com.apple.foundationdb.record.TestRecords1Proto;
-import com.apple.foundationdb.record.logging.KeyValueLogMessage;
-import com.apple.foundationdb.record.logging.LogMessageKeys;
 import com.apple.foundationdb.record.metadata.Index;
 import com.apple.foundationdb.record.metadata.IndexOptions;
 import com.apple.foundationdb.record.metadata.IndexTypes;
@@ -1158,45 +1156,4 @@ void testIndexFromIndexBlock() {
         assertReadable(tgtIndex);
         scrubAndValidate(List.of(tgtIndex));
     }
-
-    @Test
-    void testIndexFromIndexIgnoreSyncLock() {
-
-        final long numRecords = 180;
-
-        Index srcIndex = new Index("src_index", field("num_value_2"), EmptyKeyExpression.EMPTY, IndexTypes.VALUE, IndexOptions.UNIQUE_OPTIONS);
-        Index tgtIndex = new Index("tgt_index", field("num_value_3_indexed"), IndexTypes.VALUE);
-        FDBRecordStoreTestBase.RecordMetaDataHook hook = myHook(srcIndex, tgtIndex);
-
-        populateData(numRecords);
-
-        openSimpleMetaData(hook);
-        buildIndexClean(srcIndex);
-        disableAll(List.of(tgtIndex));
-
-        openSimpleMetaData(hook);
-
-        IntStream.rangeClosed(0, 4).parallel().forEach(id -> {
-            snooze(100 - id);
-            try {
-                try (OnlineIndexer indexBuilder = newIndexerBuilder(tgtIndex)
-                        .setIndexingPolicy(OnlineIndexer.IndexingPolicy.newBuilder()
-                                        .setSourceIndex("src_index")
-                                        .forbidRecordScan())
-                        .setLimit(5)
-                        .setUseSynchronizedSession(id == 0)
-                        .setMaxRetries(100) // enough to avoid giving up
-                        .build()) {
-                    indexBuilder.buildIndex(true);
-                }
-            } catch (IndexingBase.UnexpectedReadableException ex) {
-                LOGGER.info(KeyValueLogMessage.of("Ignoring lock, got exception",
-                        LogMessageKeys.SESSION_ID, id,
-                        LogMessageKeys.ERROR, ex.getMessage()));
-            }
-        });
-
-        assertReadable(List.of(tgtIndex));
-        scrubAndValidate(List.of(tgtIndex));
-    }
 }