Some cleanup

Author: jjezra

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/logging/LogMessageKeys.java

@@ -163,8 +163,7 @@ public enum LogMessageKeys {
     RECORDS_PER_SECOND,
     DOCUMENT,
     SESSION_ID,
-    EXISTING_SESSION_ID,
-    INDEXER_SESSION_ID,
+    EXISTING_INDEXER_ID,
     INDEXER_ID,
     INDEX_STATE_PRECONDITION,
     INITIAL_INDEX_STATE,

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

@@ -305,7 +305,6 @@ public CompletableFuture<Boolean> markReadableIfBuilt() {
         ).thenApply(ignore -> allReadable.get()), common.indexLogMessageKeyValues("IndexingBase::markReadableIfBuilt"));
     }
 
-
     @Nonnull
     public CompletableFuture<Boolean> markIndexReadable(boolean markReadablePlease) {
         if (!markReadablePlease) {
@@ -861,6 +860,14 @@ private CompletableFuture<Void> clearHeartbeats() {
                 .thenAccept(ignore -> heartbeat = null);
     }
 
+    private void clearHeartbeats(FDBRecordStore store) {
+        if (heartbeat != null) {
+            for (Index index : common.getTargetIndexes()) {
+                clearHeartbeatSingleTarget(store, index);
+            }
+        }
+    }
+
     private CompletableFuture<Void> clearHeartbeatSingleTarget(Index index) {
         return getRunner().runAsync(context ->
                 common.getRecordStoreBuilder().copyBuilder().setContext(context).openAsync()
@@ -1027,11 +1034,8 @@ public CompletableFuture<Void> rebuildIndexAsync(@Nonnull FDBRecordStore store)
         }))
                 .thenCompose(vignore -> setIndexingTypeOrThrow(store, false))
                 .thenCompose(vignore -> rebuildIndexInternalAsync(store))
-                .whenComplete((ignore, ignoreEx) -> {
-                    for (Index index: common.getTargetIndexes()) {
-                        clearHeartbeatSingleTarget(store, index);
-                    }
-                });
+                // If any of the indexes' heartbeats, for any reason, was not cleared during "mark readable", clear it here
+                .whenComplete((ignore, ignoreEx) ->  clearHeartbeats(store));
     }
 
     abstract CompletableFuture<Void> rebuildIndexInternalAsync(FDBRecordStore store);

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

@@ -66,6 +66,7 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
         switch (indexingMethod) {
             case SCRUB_REPAIR:
             case MUTUAL_BY_RECORDS:
+                updateHeartbeat(store, index);
                 return AsyncUtil.DONE;
 
             case BY_RECORDS:
@@ -79,7 +80,11 @@ public CompletableFuture<Void> checkAndUpdateHeartbeat(@Nonnull FDBRecordStore s
                             }
                             validateNonCompetingHeartbeat(iterator.next(), nowMilliseconds());
                             return true;
-                        }));
+                        }))
+                        .thenApply(ignore -> {
+                            updateHeartbeat(store, index);
+                            return null;
+                        });
 
             default:
                 throw new IndexingBase.ValidationException("invalid indexing method",
@@ -92,15 +97,15 @@ private void validateNonCompetingHeartbeat(KeyValue kv, long now) {
         if (keyTuple.size() < 2) { // expecting 8
             return;
         }
-        final UUID otherSessionId = keyTuple.getUUID(keyTuple.size() - 1);
-        if (!otherSessionId.equals(this.indexerId)) {
+        final UUID otherIndexerId = keyTuple.getUUID(keyTuple.size() - 1);
+        if (!otherIndexerId.equals(this.indexerId)) {
             try {
                 final IndexBuildProto.IndexBuildHeartbeat otherHeartbeat = IndexBuildProto.IndexBuildHeartbeat.parseFrom(kv.getValue());
                 final long age = now - otherHeartbeat.getHeartbeatTimeMilliseconds();
                 if (age > 0 && age < leaseLength) {
                     throw new SynchronizedSessionLockedException("Failed to initialize the session because of an existing session in progress")
-                            .addLogInfo(LogMessageKeys.SESSION_ID, indexerId)
-                            .addLogInfo(LogMessageKeys.EXISTING_SESSION_ID, otherSessionId)
+                            .addLogInfo(LogMessageKeys.INDEXER_ID, indexerId)
+                            .addLogInfo(LogMessageKeys.EXISTING_INDEXER_ID, otherIndexerId)
                             .addLogInfo(LogMessageKeys.AGE_MILLISECONDS, age)
                             .addLogInfo(LogMessageKeys.TIME_LIMIT_MILLIS, leaseLength);
                 }
@@ -131,13 +136,13 @@ public static CompletableFuture<Map<UUID, IndexBuildProto.IndexBuildHeartbeat>>
                             if (keyTuple.size() < 2) { // expecting 8
                                 return true; // ignore, next
                             }
-                            final UUID otherSessionId = keyTuple.getUUID(keyTuple.size() - 1);
+                            final UUID otherIndexerId = keyTuple.getUUID(keyTuple.size() - 1);
                             try {
                                 final IndexBuildProto.IndexBuildHeartbeat otherHeartbeat = IndexBuildProto.IndexBuildHeartbeat.parseFrom(kv.getValue());
-                                ret.put(otherSessionId, otherHeartbeat);
+                                ret.put(otherIndexerId, otherHeartbeat);
                             } catch (InvalidProtocolBufferException e) {
                                 // put a NONE heartbeat to indicate an invalid item
-                                ret.put(otherSessionId, IndexBuildProto.IndexBuildHeartbeat.newBuilder()
+                                ret.put(otherIndexerId, IndexBuildProto.IndexBuildHeartbeat.newBuilder()
                                         .setMethod(IndexBuildProto.IndexBuildIndexingStamp.Method.NONE)
                                         .build());
                             }

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

@@ -100,12 +100,12 @@ public static Subspace indexheartbeatSubspace(@Nonnull FDBRecordStoreBase<?> sto
      * Subspace that stores the indexing heartbeat.
      * @param store store
      * @param index index
-     * @param sessionId session id
+     * @param indexerId session id
      * @return subspace
      */
     @Nonnull
-    public static Subspace indexheartbeatSubspace(@Nonnull FDBRecordStoreBase<?> store, @Nonnull Index index, @Nonnull UUID sessionId) {
-        return indexheartbeatSubspace(store, index).subspace(Tuple.from(sessionId));
+    public static Subspace indexheartbeatSubspace(@Nonnull FDBRecordStoreBase<?> store, @Nonnull Index index, @Nonnull UUID indexerId) {
+        return indexheartbeatSubspace(store, index).subspace(Tuple.from(indexerId));
     }
 
     /**
@@ -209,7 +209,7 @@ public static void eraseAllIndexingDataButTheLock(@Nonnull FDBRecordContext cont
         eraseAllIndexingScrubbingData(context, store, index);
         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 the clearing within the same transaction.
+        // 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()));
     }
 }

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

@@ -1267,7 +1267,7 @@ public static class Builder {
             private DesiredAction ifReadable = DesiredAction.CONTINUE;
             private boolean doAllowUniquePendingState = false;
             private Set<TakeoverTypes> allowedTakeoverSet = null;
-            private long checkIndexingStampFrequency = 30_000;
+            private long checkIndexingStampFrequency = 10_000;
             private boolean useMutualIndexing = false;
             private List<Tuple> useMutualIndexingBoundaries = null;
             private boolean allowUnblock = false;

fdb-record-layer-core/src/main/proto/index_build.proto

@@ -36,7 +36,6 @@ message IndexBuildIndexingStamp {
     MUTUAL_BY_RECORDS = 5; // scan records, build multiple target indexes, while allowing multiple indexer processes
     NONE = 6; // return this stamp in query when null - never to be written in the DB
   };
-
   optional Method method = 1;
   optional bytes source_index_subspace_key = 2; // relevant only with BY_INDEX method
   optional int32 source_index_last_modified_version = 3; // only with BY_INDEX method