Apply (more of) Scott's requested changes

Author: jjezra

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

@@ -4965,6 +4965,7 @@ public void removeFormerIndex(FormerIndex formerIndex) {
     }
 
     private void clearReadableIndexBuildData(Index index) {
+        // Clear index maintenance data that is unneeded once the index becomes readable
         IndexingRangeSet.forIndexBuild(this, index).clear();
         IndexingHeartbeat.clearAllHeartbeats(this, index);
     }

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

@@ -102,6 +102,7 @@ public abstract class IndexingBase {
     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
 
     IndexingBase(@Nonnull IndexingCommon common,
@@ -157,15 +158,13 @@ public CompletableFuture<Void> buildIndexAsync(boolean markReadable) {
         long startNanos = System.nanoTime();
         FDBDatabaseRunner runner = getRunner();
         final FDBStoreTimer timer = runner.getTimer();
-        if ( timer != null) {
+        if (timer != null) {
             lastProgressSnapshot = StoreTimerSnapshot.from(timer);
         }
-        AtomicReference<Throwable> indexingException = new AtomicReference<>(null);
-        return handleStateAndDoBuildIndexAsync(markReadable, message)
-                .handle((ret, ex) -> {
-                    if (ex != null) {
-                        indexingException.set(ex);
-                    }
+        return MoreAsyncUtil.composeWhenComplete(
+                handleStateAndDoBuildIndexAsync(markReadable, message),
+                (result, ex) -> {
+                    // proper log
                     message.addKeysAndValues(indexingLogMessageKeyValues()) // add these here to pick up state accumulated during build
                             .addKeysAndValues(common.indexLogMessageKeyValues())
                             .addKeyAndValue(LogMessageKeys.TOTAL_MICROS, TimeUnit.NANOSECONDS.toMicros(System.nanoTime() - startNanos));
@@ -177,20 +176,13 @@ public CompletableFuture<Void> buildIndexAsync(boolean markReadable) {
                         message.addKeyAndValue(LogMessageKeys.RESULT, "success");
                         LOGGER.info(message.toString());
                     }
-                    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();
-                    if (ex instanceof RuntimeException) {
-                        throw (RuntimeException) ex;
-                    } else if (ex != null) {
-                        throw new RuntimeException(ex);
-                    }
-                    return null;
-                });
+                    // 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.
+                    return clearHeartbeats()
+                            .handle((ignoreRet, ignoreEx) -> null);
+                },
+                getRunner().getDatabase()::mapAsyncToSyncException
+        );
     }
 
     abstract List<Object> indexingLogMessageKeyValues();
@@ -271,7 +263,8 @@ private CompletableFuture<Void> handleStateAndDoBuildIndexAsync(boolean markRead
                 doIndex ?
                 buildIndexInternalAsync().thenApply(ignore -> markReadable) :
                 AsyncUtil.READY_FALSE
-        ).thenCompose(this::markIndexReadable).thenApply(ignore -> null);
+        ).thenCompose(this::markIndexReadable
+        ).thenApply(ignore -> null);
     }
 
     private CompletableFuture<Void> markIndexesWriteOnly(boolean continueBuild, FDBRecordStore store) {
@@ -317,7 +310,7 @@ public CompletableFuture<Boolean> markIndexReadable(boolean markReadablePlease)
         // Mark each index readable in its own (retriable, parallel) transaction. If one target fails to become
         // readable, it should not affect the others.
         return forEachTargetIndex(index ->
-                markIndexReadableSingleTarget(index, anythingChanged, runtimeExceptionAtomicReference)
+                markIndexReadableForIndex(index, anythingChanged, runtimeExceptionAtomicReference)
         ).thenApply(ignore -> {
             RuntimeException ex = runtimeExceptionAtomicReference.get();
             if (ex != null) {
@@ -328,13 +321,13 @@ public CompletableFuture<Boolean> markIndexReadable(boolean markReadablePlease)
         });
     }
 
-    private CompletableFuture<Boolean> markIndexReadableSingleTarget(Index index, AtomicBoolean anythingChanged,
-                                                                     AtomicReference<RuntimeException> runtimeExceptionAtomicReference) {
+    private CompletableFuture<Boolean> markIndexReadableForIndex(Index index, AtomicBoolean anythingChanged,
+                                                                 AtomicReference<RuntimeException> runtimeExceptionAtomicReference) {
         // An extension function to reduce markIndexReadable's complexity
         return getRunner().runAsync(context ->
                 common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
                         .thenCompose(store -> {
-                            clearHeartbeatSingleTarget(store, index);
+                            clearHeartbeatForIndex(store, index);
                             return policy.shouldAllowUniquePendingState(store) ?
                                    store.markIndexReadableOrUniquePending(index) :
                                    store.markIndexReadable(index);
@@ -376,7 +369,7 @@ private CompletableFuture<Void> setIndexingTypeOrThrow(FDBRecordStore store, boo
         if (forceStampOverwrite && !continuedBuild) {
             // Fresh session + overwrite = no questions asked
             store.saveIndexingTypeStamp(index, newStamp);
-            return AsyncUtil.DONE ;
+            return AsyncUtil.DONE;
         }
         return store.loadIndexingTypeStampAsync(index)
                 .thenCompose(savedStamp -> {
@@ -858,30 +851,27 @@ private CompletableFuture<Void> updateHeartbeat(boolean validate, FDBRecordStore
 
     private CompletableFuture<Void> clearHeartbeats() {
         if (heartbeat == null) {
+            // Here: either silent heartbeats or heartbeats had been cleared during markReadable phase
             return AsyncUtil.DONE;
         }
-        return forEachTargetIndex(this::clearHeartbeatSingleTarget)
-                .thenAccept(ignore -> heartbeat = null);
+        // Here: for each index we clear (only) the heartbeat generated by this indexer. This is a quick operation that can be done in a single transaction.
+        return getRunner().runAsync(context ->
+                common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
+                        .thenApply(store -> {
+                            clearHeartbeats(store);
+                            return null;
+                        }));
     }
 
     private void clearHeartbeats(FDBRecordStore store) {
         if (heartbeat != null) {
             for (Index index : common.getTargetIndexes()) {
-                clearHeartbeatSingleTarget(store, index);
+                heartbeat.clearHeartbeat(store, index);
             }
         }
     }
 
-    private CompletableFuture<Void> clearHeartbeatSingleTarget(Index index) {
-        return getRunner().runAsync(context ->
-                common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
-                        .thenApply(store -> {
-                            clearHeartbeatSingleTarget(store, index);
-                            return null;
-                        }));
-    }
-
-    private void clearHeartbeatSingleTarget(FDBRecordStore store, Index index) {
+    private void clearHeartbeatForIndex(FDBRecordStore store, Index index) {
         if (heartbeat != null) {
             heartbeat.clearHeartbeat(store, index);
         }
@@ -1115,9 +1105,9 @@ public CompletableFuture<Map<UUID, IndexBuildProto.IndexBuildHeartbeat>> getInde
                         .thenCompose(store -> IndexingHeartbeat.getIndexingHeartbeats(store, common.getPrimaryIndex(), maxCount)));
     }
 
-    public CompletableFuture<Integer> clearIndexingHeartbeats(long minAgenMilliseconds, int maxIteration) {
+    public CompletableFuture<Integer> clearIndexingHeartbeats(long minAgeMilliseconds, int maxIteration) {
         return getRunner().runAsync(context -> openRecordStore(context)
-                .thenCompose(store -> IndexingHeartbeat.clearIndexingHeartbeats(store, common.getPrimaryIndex(), minAgenMilliseconds, maxIteration)));
+                .thenCompose(store -> IndexingHeartbeat.clearIndexingHeartbeats(store, common.getPrimaryIndex(), minAgeMilliseconds, maxIteration)));
     }
 
     /**

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

@@ -21,13 +21,17 @@
 package com.apple.foundationdb.record.provider.foundationdb;
 
 import com.apple.foundationdb.KeyValue;
+import com.apple.foundationdb.annotation.API;
 import com.apple.foundationdb.async.AsyncIterator;
 import com.apple.foundationdb.async.AsyncUtil;
 import com.apple.foundationdb.record.IndexBuildProto;
+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.synchronizedsession.SynchronizedSessionLockedException;
 import com.google.protobuf.InvalidProtocolBufferException;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nonnull;
 import java.util.HashMap;
@@ -36,8 +40,20 @@
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicInteger;
 
+/**
+ * Indexing Shared Heartbeats can be used to define and handle "active" indexing processes.
+ * Every indexer should update its unique heartbeat during its indexing iteration. If the indexing session is optimized for
+ * non-mutual (as defined by the indexing type, see {@link IndexBuildProto.IndexBuildIndexingStamp}), detecting an existing
+ * active heartbeat will help preventing concurrent, conflicting, indexing attempts.
+ * In addition, the heartbeats can be used by users to query activity status of ongoing indexing sessions.
+ */
+@API(API.Status.INTERNAL)
 public class IndexingHeartbeat {
     // [prefix, indexerId] -> [indexing-type, genesis time, heartbeat time]
+    @Nonnull
+    private static final Logger logger = LoggerFactory.getLogger(IndexingHeartbeat.class);
+    public static final String INVALID_HEARTBEAT_INFO = "<< Invalid Heartbeat >>";
+
     final UUID indexerId;
     final String info;
     final long genesisTimeMilliseconds;
@@ -53,7 +69,7 @@ public IndexingHeartbeat(final UUID indexerId, String info, long leaseLength, bo
     }
 
     public void updateHeartbeat(@Nonnull FDBRecordStore store, @Nonnull Index index) {
-        byte[] key = IndexingSubspaces.indexheartbeatSubspace(store, index, indexerId).pack();
+        byte[] key = IndexingSubspaces.indexHeartbeatSubspace(store, index, indexerId).pack();
         byte[] value = IndexBuildProto.IndexBuildHeartbeat.newBuilder()
                 .setInfo(info)
                 .setGenesisTimeMilliseconds(genesisTimeMilliseconds)
@@ -92,7 +108,11 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
                                     }
                                 }
                             } catch (InvalidProtocolBufferException e) {
-                                throw new RuntimeException(e);
+                                if (logger.isWarnEnabled()) {
+                                    logger.warn(KeyValueLogMessage.of("Bad indexing heartbeat item",
+                                            LogMessageKeys.KEY, kv.getKey(),
+                                            LogMessageKeys.VALUE, kv.getValue()));
+                                }
                             }
                             return true;
                         }))
@@ -103,11 +123,11 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
     }
 
     public void clearHeartbeat(@Nonnull FDBRecordStore store, @Nonnull Index index) {
-        store.ensureContextActive().clear(IndexingSubspaces.indexheartbeatSubspace(store, index, indexerId).pack());
+        store.ensureContextActive().clear(IndexingSubspaces.indexHeartbeatSubspace(store, index, indexerId).pack());
     }
 
     public static void clearAllHeartbeats(@Nonnull FDBRecordStore store, @Nonnull Index index) {
-        store.ensureContextActive().clear(IndexingSubspaces.indexheartbeatSubspace(store, index).range());
+        store.ensureContextActive().clear(IndexingSubspaces.indexHeartbeatSubspace(store, index).range());
     }
 
     public static CompletableFuture<Map<UUID, IndexBuildProto.IndexBuildHeartbeat>> getIndexingHeartbeats(FDBRecordStore store, Index index, int maxCount) {
@@ -130,15 +150,17 @@ public static CompletableFuture<Map<UUID, IndexBuildProto.IndexBuildHeartbeat>>
                             } catch (InvalidProtocolBufferException e) {
                                 // Let the caller know about this invalid heartbeat.
                                 ret.put(otherIndexerId, IndexBuildProto.IndexBuildHeartbeat.newBuilder()
-                                        .setInfo("<< Invalid Heartbeat >>")
+                                        .setInfo(INVALID_HEARTBEAT_INFO)
+                                                .setGenesisTimeMilliseconds(0)
+                                                .setHeartbeatTimeMilliseconds(0)
                                         .build());
                             }
                             return true;
                         }))
                 .thenApply(ignore -> ret);
     }
 
-    public static CompletableFuture<Integer> clearIndexingHeartbeats(@Nonnull FDBRecordStore store, @Nonnull Index index, long minAgenMilliseconds, int maxIteration) {
+    public static CompletableFuture<Integer> clearIndexingHeartbeats(@Nonnull FDBRecordStore store, @Nonnull Index index, long minAgeMilliseconds, int maxIteration) {
         final AsyncIterator<KeyValue> iterator = heartbeatsIterator(store, index);
         final AtomicInteger deleteCount = new AtomicInteger(0);
         final AtomicInteger iterationCount = new AtomicInteger(0);
@@ -156,7 +178,7 @@ public static CompletableFuture<Integer> clearIndexingHeartbeats(@Nonnull FDBRec
                             try {
                                 final IndexBuildProto.IndexBuildHeartbeat otherHeartbeat = IndexBuildProto.IndexBuildHeartbeat.parseFrom(kv.getValue());
                                 // remove heartbeat if too old
-                                shouldRemove = now + minAgenMilliseconds >= otherHeartbeat.getHeartbeatTimeMilliseconds();
+                                shouldRemove = now >= otherHeartbeat.getHeartbeatTimeMilliseconds() +  minAgeMilliseconds;
                             } catch (InvalidProtocolBufferException e) {
                                 // remove heartbeat if invalid
                                 shouldRemove = true;
@@ -171,11 +193,11 @@ public static CompletableFuture<Integer> clearIndexingHeartbeats(@Nonnull FDBRec
     }
 
     private static AsyncIterator<KeyValue> heartbeatsIterator(FDBRecordStore store, Index index) {
-        return store.getContext().ensureActive().snapshot().getRange(IndexingSubspaces.indexheartbeatSubspace(store, index).range()).iterator();
+        return store.getContext().ensureActive().getRange(IndexingSubspaces.indexHeartbeatSubspace(store, index).range()).iterator();
     }
 
     private static UUID heartbeatKeyToIndexerId(FDBRecordStore store, Index index, byte[] key) {
-        return IndexingSubspaces.indexheartbeatSubspace(store, index).unpack(key).getUUID(0);
+        return IndexingSubspaces.indexHeartbeatSubspace(store, index).unpack(key).getUUID(0);
     }
 
     private static long nowMilliseconds() {

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

@@ -92,7 +92,7 @@ public static Subspace indexBuildTypeSubspace(@Nonnull FDBRecordStoreBase<?> sto
      * @return subspace
      */
     @Nonnull
-    public static Subspace indexheartbeatSubspace(@Nonnull FDBRecordStoreBase<?> store, @Nonnull Index index) {
+    public static Subspace indexHeartbeatSubspace(@Nonnull FDBRecordStoreBase<?> store, @Nonnull Index index) {
         return indexBuildSubspace(store, index, INDEX_BUILD_HEARTBEAT_PREFIX);
     }
 
@@ -104,8 +104,8 @@ public static Subspace indexheartbeatSubspace(@Nonnull FDBRecordStoreBase<?> sto
      * @return subspace
      */
     @Nonnull
-    public static Subspace indexheartbeatSubspace(@Nonnull FDBRecordStoreBase<?> store, @Nonnull Index index, @Nonnull UUID indexerId) {
-        return indexheartbeatSubspace(store, index).subspace(Tuple.from(indexerId));
+    public static Subspace indexHeartbeatSubspace(@Nonnull FDBRecordStoreBase<?> store, @Nonnull Index index, @Nonnull UUID indexerId) {
+        return indexHeartbeatSubspace(store, index).subspace(Tuple.from(indexerId));
     }
 
     /**
@@ -210,6 +210,6 @@ public static void eraseAllIndexingDataButTheLock(@Nonnull FDBRecordContext cont
         context.clear(Range.startsWith(indexBuildScannedRecordsSubspace(store, index).pack()));
         context.clear(Range.startsWith(indexBuildTypeSubspace(store, index).pack()));
         // The heartbeats, unlike the sync lock, may be erased here. If needed, an appropriate heartbeat will be set after this clear & within the same transaction.
-        context.clear(Range.startsWith(indexheartbeatSubspace(store, index).pack()));
+        context.clear(Range.startsWith(indexHeartbeatSubspace(store, index).pack()));
     }
 }