Apply (even more of) Scott's requested changes

Author: jjezra

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

@@ -100,7 +100,6 @@ public abstract class IndexingBase {
     private StoreTimerSnapshot lastProgressSnapshot = null;
     private boolean forceStampOverwrite = false;
     private final long startingTimeMillis;
-    private long lastTypeStampCheckMillis;
     private Map<String, IndexingMerger> indexingMergerMap = null;
     @Nullable
     private IndexingHeartbeat heartbeat = null; // this will stay null for index scrubbing
@@ -119,7 +118,6 @@ public abstract class IndexingBase {
         this.isScrubber = isScrubber;
         this.throttle = new IndexingThrottle(common, isScrubber);
         this.startingTimeMillis = System.currentTimeMillis();
-        this.lastTypeStampCheckMillis = startingTimeMillis;
     }
 
     // helper functions
@@ -176,8 +174,9 @@ public CompletableFuture<Void> buildIndexAsync(boolean markReadable) {
                         message.addKeyAndValue(LogMessageKeys.RESULT, "success");
                         LOGGER.info(message.toString());
                     }
-                    // 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.
+                    // Here: if the heartbeats were not fully cleared while marking the index as readable, they will be cleared in
+                    // this dedicated transaction. Clearing the heartbeats at the end of the indexing session is a "best effort"
+                    // operation, hence exceptions are ignored.
                     return clearHeartbeats()
                             .handle((ignoreRet, ignoreEx) -> null);
                 },
@@ -825,18 +824,17 @@ private CompletableFuture<Boolean> hadTransactionReachedLimits(FDBRecordStore st
     }
 
     private CompletableFuture<Void> validateTypeStamp(@Nonnull FDBRecordStore store) {
-        if (shouldValidate()) {
-            // check other heartbeats (if exclusive) & typestamp
-            final IndexBuildProto.IndexBuildIndexingStamp expectedTypeStamp = getIndexingTypeStamp(store);
-            return forEachTargetIndex(index -> CompletableFuture.allOf(
-                    updateHeartbeat(true, store, index),
-                    store.loadIndexingTypeStampAsync(index)
-                            .thenAccept(typeStamp -> validateTypeStamp(typeStamp, expectedTypeStamp, index))
-                    ));
-        } else {
-            // update only
-            return forEachTargetIndex(index -> updateHeartbeat(false, store, index));
+        // check other heartbeats (if exclusive) & typestamp
+        if (isScrubber) {
+            // Scrubber's type-stamp is never commited. It is protected by expecting a READABLE index state.
+            return AsyncUtil.DONE;
         }
+        final IndexBuildProto.IndexBuildIndexingStamp expectedTypeStamp = getIndexingTypeStamp(store);
+        return forEachTargetIndex(index -> CompletableFuture.allOf(
+                updateHeartbeat(true, store, index),
+                store.loadIndexingTypeStampAsync(index)
+                        .thenAccept(typeStamp -> validateTypeStamp(typeStamp, expectedTypeStamp, index))
+        ));
     }
 
     private CompletableFuture<Void> updateHeartbeat(boolean validate, FDBRecordStore store, Index index) {
@@ -877,22 +875,6 @@ private void clearHeartbeatForIndex(FDBRecordStore store, Index index) {
         }
     }
 
-
-    private boolean shouldValidate() {
-        final long minimalInterval = policy.getCheckIndexingMethodFrequencyMilliseconds();
-        if (minimalInterval < 0 || isScrubber) {
-            return false;
-        }
-        if (minimalInterval > 0) {
-            final long now = System.currentTimeMillis();
-            if (now < lastTypeStampCheckMillis + minimalInterval) {
-                return false;
-            }
-            lastTypeStampCheckMillis = now;
-        }
-        return true;
-    }
-
     private void validateTypeStamp(final IndexBuildProto.IndexBuildIndexingStamp typeStamp,
                                    final IndexBuildProto.IndexBuildIndexingStamp expectedTypeStamp,
                                    Index index) {
@@ -1028,7 +1010,6 @@ public CompletableFuture<Void> rebuildIndexAsync(@Nonnull FDBRecordStore store)
         }))
                 .thenCompose(vignore -> setIndexingTypeOrThrow(store, false))
                 .thenCompose(vignore -> rebuildIndexInternalAsync(store))
-                // If any of the indexes' heartbeats, for any reason, was not cleared during "mark readable", clear it here
                 .whenComplete((ignore, ignoreEx) ->  clearHeartbeats(store));
     }
 

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

@@ -93,33 +93,34 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
                                 return false;
                             }
                             final KeyValue kv = iterator.next();
-                            try {
-                                final UUID otherIndexerId = heartbeatKeyToIndexerId(store, index, kv.getKey());
-                                if (!otherIndexerId.equals(this.indexerId)) {
-                                    final IndexBuildProto.IndexBuildHeartbeat otherHeartbeat = IndexBuildProto.IndexBuildHeartbeat.parseFrom(kv.getValue());
-                                    final long age = now - otherHeartbeat.getHeartbeatTimeMilliseconds();
-                                    if (age > 0 && age < leaseLength) {
-                                        // For practical reasons, this exception is backward compatible to the Synchronized Lock one
-                                        throw new SynchronizedSessionLockedException("Failed to initialize the session because of an existing session in progress")
-                                                .addLogInfo(LogMessageKeys.INDEXER_ID, indexerId)
-                                                .addLogInfo(LogMessageKeys.EXISTING_INDEXER_ID, otherIndexerId)
-                                                .addLogInfo(LogMessageKeys.AGE_MILLISECONDS, age)
-                                                .addLogInfo(LogMessageKeys.TIME_LIMIT_MILLIS, leaseLength);
-                                    }
-                                }
-                            } catch (InvalidProtocolBufferException e) {
-                                if (logger.isWarnEnabled()) {
-                                    logger.warn(KeyValueLogMessage.of("Bad indexing heartbeat item",
-                                            LogMessageKeys.KEY, kv.getKey(),
-                                            LogMessageKeys.VALUE, kv.getValue()));
-                                }
-                            }
+                            checkSingleHeartbeat(store, index, kv, now);
                             return true;
                         }))
-                .thenApply(ignore -> {
-                    updateHeartbeat(store, index);
-                    return null;
-                });
+                .thenAccept(ignore -> updateHeartbeat(store, index));
+    }
+
+    private void checkSingleHeartbeat(final @Nonnull FDBRecordStore store, final @Nonnull Index index, final KeyValue kv, final long now) {
+        try {
+            final UUID otherIndexerId = heartbeatKeyToIndexerId(store, index, kv.getKey());
+            if (!otherIndexerId.equals(this.indexerId)) {
+                final IndexBuildProto.IndexBuildHeartbeat otherHeartbeat = IndexBuildProto.IndexBuildHeartbeat.parseFrom(kv.getValue());
+                final long age = now - otherHeartbeat.getHeartbeatTimeMilliseconds();
+                if (age > 0 && age < leaseLength) {
+                    // For practical reasons, this exception is backward compatible to the Synchronized Lock one
+                    throw new SynchronizedSessionLockedException("Failed to initialize the session because of an existing session in progress")
+                            .addLogInfo(LogMessageKeys.INDEXER_ID, indexerId)
+                            .addLogInfo(LogMessageKeys.EXISTING_INDEXER_ID, otherIndexerId)
+                            .addLogInfo(LogMessageKeys.AGE_MILLISECONDS, age)
+                            .addLogInfo(LogMessageKeys.TIME_LIMIT_MILLIS, leaseLength);
+                }
+            }
+        } catch (InvalidProtocolBufferException e) {
+            if (logger.isWarnEnabled()) {
+                logger.warn(KeyValueLogMessage.of("Bad indexing heartbeat item",
+                        LogMessageKeys.KEY, kv.getKey(),
+                        LogMessageKeys.VALUE, kv.getValue()));
+            }
+        }
     }
 
     public void clearHeartbeat(@Nonnull FDBRecordStore store, @Nonnull Index index) {

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

@@ -315,11 +315,11 @@ private IndexingBase getIndexer() {
     }
 
     /**
+     * <em>Deprecated</em> and unused.
      * This {@link Exception} can be thrown in the case that one calls one of the methods
      * that explicitly state that they are building an unbuilt range, i.e., a range of keys
      * that contains no keys which have yet been processed by the {@link OnlineIndexer}
      * during an index build.
-     * <em>Deprecated</em> and unused.
      */
     @API(API.Status.DEPRECATED)
     @SuppressWarnings("serial")
@@ -480,7 +480,7 @@ public boolean checkAnyOngoingOnlineIndexBuilds() {
     }
 
     /**
-     * Check if the main index is being built. Note that with shared heartbeats, this function will now return true for any active session - mutual or exclusive.
+     * Check if the provided index is being built. Note that with shared heartbeats, this function will now return true for any active session - mutual or exclusive.
      * @return a future that will complete to <code>true</code> if the index is being built and <code>false</code> otherwise
      */
     public CompletableFuture<Boolean> checkAnyOngoingOnlineIndexBuildsAsync() {
@@ -490,7 +490,7 @@ public CompletableFuture<Boolean> checkAnyOngoingOnlineIndexBuildsAsync() {
     }
 
     /**
-     * Check if the main index is being built. Note that with shared heartbeats, this function will now return true for any active session - mutual or exclusive.
+     * Check if the provided index is being built. Note that with shared heartbeats, this function will now return true for any active session - mutual or exclusive.
      * Where "active session" is determined by an indexing heartbeat that is less than {@link OnlineIndexOperationConfig#DEFAULT_LEASE_LENGTH_MILLIS} old.
      * @param recordStore record store whose index builds need to be checked
      * @param index the index to check for ongoing index builds
@@ -501,7 +501,7 @@ public static CompletableFuture<Boolean> checkAnyOngoingOnlineIndexBuildsAsync(@
     }
 
     /**
-     * Check if the main index is being built. Note that with shared heartbeats, this function will now return true for any active session - mutual or exclusive.
+     * Check if the provided index is being built. Note that with shared heartbeats, this function will now return true for any active session - mutual or exclusive.
      * @param recordStore record store whose index builds need to be checked
      * @param index the index to check for ongoing index builds
      * @param leasingMilliseconds max heartbeat age to be considered an "active session"
@@ -517,7 +517,14 @@ public static CompletableFuture<Boolean> checkAnyOngoingOnlineIndexBuildsAsync(@
 
     /**
      * Builds an index across multiple transactions.
-     * This is a slow and retrying operation that is intended to be executed by background processes.
+     * <p>
+     * If the indexing session is not mutual, it will stop with {@link com.apple.foundationdb.synchronizedsession.SynchronizedSessionLockedException}
+     * if there is another active indexing session on the same index. It first checks and updates index states and
+     * clear index data respecting the {@link IndexStatePrecondition} being set. It then builds the index across
+     * multiple transactions honoring the rate-limiting parameters set in the constructor of this class. It also retries
+     * any retriable errors that it encounters while it runs the build. At the end, it marks the index readable in the
+     * store.
+     * </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.
@@ -586,9 +593,10 @@ public Map<String, IndexBuildProto.IndexBuildIndexingStamp> queryIndexingStamps(
 
     /**
      * Block partly built indexes, preventing continuation.
-     * Active indexing sessions will check for this block according to {@link IndexingPolicy.Builder#checkIndexingStampFrequencyMilliseconds(long)}.}
-     * @param id if non null, will be added to the "indexing stamp" as an id/hint for the blocking reason.
-     * @param ttlSeconds if non null, the block will automatically expire after this value (in seconds).
+     * Active indexing sessions will check for this block during every iterating transaction, which still requires the caller to
+     * wait a few seconds before assuming that all indexing had stopped.
+     * @param id if non-null, will be added to the "indexing stamp" as an id/hint for the blocking reason.
+     * @param ttlSeconds if non-null, the block will automatically expire after this value (in seconds).
      * @return a map of target indexes and their "indexing stamps" after the change.
      */
     @API(API.Status.EXPERIMENTAL)
@@ -635,7 +643,7 @@ public Map<UUID, IndexBuildProto.IndexBuildHeartbeat> getIndexingHeartbeats(int
      * Clear old indexing heartbeats for a given index (single target or primary index).
      * Typically, heartbeats are deleted either at the end of an indexing sessions or when the index becomes readable. This
      * cleanup function can be used if, for any reason, the heartbeats could not be deleted from the database at the end of a session.
-     * @param minAgeMilliseconds minimum heartbeat age (time elapsed since heartbeat creation, in milliseconds) to clear.
+     * @param minAgeMilliseconds minimum heartbeat age (time elapsed since the last heartbeat, in milliseconds) to clear.
      * @param maxIteration safety valve to limit number of items to check. Typically set to zero to keep unlimited
      * @return number of cleared heartbeats
      */
@@ -956,7 +964,6 @@ public static class IndexingPolicy {
         private final DesiredAction ifReadable;
         private final boolean allowUniquePendingState;
         private final Set<TakeoverTypes> allowedTakeoverSet;
-        private final long checkIndexingMethodFrequencyMilliseconds;
         private final boolean mutualIndexing;
         private final List<Tuple> mutualIndexingBoundaries;
         private final boolean allowUnblock;
@@ -1015,7 +1022,6 @@ public enum TakeoverTypes {
         private IndexingPolicy(@Nullable String sourceIndex, @Nullable Object sourceIndexSubspaceKey, boolean forbidRecordScan,
                                DesiredAction ifDisabled, DesiredAction ifWriteOnly, DesiredAction ifMismatchPrevious, DesiredAction ifReadable,
                                boolean allowUniquePendingState, Set<TakeoverTypes> allowedTakeoverSet,
-                               long checkIndexingMethodFrequencyMilliseconds,
                                boolean mutualIndexing, List<Tuple> mutualIndexingBoundaries,
                                boolean allowUnblock, String allowUnblockId,
                                long initialMergesCountLimit,
@@ -1029,7 +1035,6 @@ private IndexingPolicy(@Nullable String sourceIndex, @Nullable Object sourceInde
             this.ifReadable = ifReadable;
             this.allowUniquePendingState = allowUniquePendingState;
             this.allowedTakeoverSet = allowedTakeoverSet;
-            this.checkIndexingMethodFrequencyMilliseconds = checkIndexingMethodFrequencyMilliseconds;
             this.mutualIndexing = mutualIndexing;
             this.mutualIndexingBoundaries = mutualIndexingBoundaries;
             this.allowUnblock = allowUnblock;
@@ -1101,7 +1106,6 @@ public Builder toBuilder() {
                     .setIfReadable(ifReadable)
                     .allowUniquePendingState(allowUniquePendingState)
                     .allowTakeoverContinue(allowedTakeoverSet)
-                    .checkIndexingStampFrequencyMilliseconds(checkIndexingMethodFrequencyMilliseconds)
                     .setMutualIndexing(mutualIndexing)
                     .setMutualIndexingBoundaries(mutualIndexingBoundaries)
                     .setAllowUnblock(allowUnblock, allowUnblockId)
@@ -1226,11 +1230,13 @@ public boolean shouldAllowUnblock(String stampBlockId) {
         }
 
         /**
+         * <em>Deprecated</em> and unused.
          * If negative, avoid checks. Else, minimal interval between checks.
-         * @return minmal interval in milliseconds.
+         * @return minimal interval in milliseconds.
          */
+        @API(API.Status.DEPRECATED)
         public long getCheckIndexingMethodFrequencyMilliseconds() {
-            return this.checkIndexingMethodFrequencyMilliseconds;
+            return 0;
         }
 
         /**
@@ -1275,7 +1281,6 @@ public static class Builder {
             private DesiredAction ifReadable = DesiredAction.CONTINUE;
             private boolean doAllowUniquePendingState = false;
             private Set<TakeoverTypes> allowedTakeoverSet = null;
-            private long checkIndexingStampFrequency = 5_000;
             private boolean useMutualIndexing = false;
             private List<Tuple> useMutualIndexingBoundaries = null;
             private boolean allowUnblock = false;
@@ -1457,6 +1462,7 @@ public Builder allowTakeoverContinue(@Nullable Collection<TakeoverTypes> allowed
             }
 
             /**
+             * <em>Deprecated</em> - for better consistency, the type stamp will now be validated during every iterating transaction.
              * During indexing, the indexer can check the current indexing stamp and throw an exception if it had changed.
              * This may happen if another indexing type takes over or by an indexing block (see {@link #indexingStamp}).
              * The argument may be:
@@ -1467,8 +1473,9 @@ public Builder allowTakeoverContinue(@Nullable Collection<TakeoverTypes> allowed
              * @param frequency : If negative, avoid checks. Else, minimal interval between checks
              * @return this builder.
              */
+            @API(API.Status.DEPRECATED)
             public Builder checkIndexingStampFrequencyMilliseconds(long frequency) {
-                this.checkIndexingStampFrequency = frequency;
+                // No-op
                 return this;
             }
 
@@ -1577,7 +1584,6 @@ public IndexingPolicy build() {
                 return new IndexingPolicy(sourceIndex, sourceIndexSubspaceKey, forbidRecordScan,
                         ifDisabled, ifWriteOnly, ifMismatchPrevious, ifReadable,
                         doAllowUniquePendingState, allowedTakeoverSet,
-                        checkIndexingStampFrequency,
                         useMutualIndexing, useMutualIndexingBoundaries, allowUnblock, allowUnblockId,
                         initialMergesCountLimit, reverseScanOrder);
             }