Create throttled and retrying record iterator (#3350)

This PR creates the `ThrottledRetryingIterator` for use with cases that
need a reliable retrying mechanism with resource controls.
for the most part, this is new code that is not in use yet. Small
changes to the existing code base (log message fields, executor method
made public, refactoring of the FutureAutoClose) were made as well.
This class is meant to live at the same level as other
`FDBDatabaseRunner`s even though this class does not implement this
interface. Hence, this is using the `TransactionalRunner` and
`FutureAutoClose` classes instead of the (already retrying)
`FDBDatabaseRunnerImpl`.
The iterator implements resource controls - ops/sec and ops/transaction
for scan and delete operations. It does not implement similar controls
for other operations (update, create etc). These can be added later when
they are needed.

Resolves #3355

Author: ohadzeliger

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

@@ -312,6 +312,7 @@ public enum LogMessageKeys {
     TOTAL_RECORDS_SCANNED,
     TOTAL_RECORDS_SCANNED_DURING_FAILURES,
     SCRUB_TYPE,
+    RETRY_COUNT,
 
     // time limits milliseconds
     TIME_LIMIT_MILLIS("time_limit_milliseconds"),

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

@@ -740,7 +740,13 @@ public Executor getExecutor() {
         return factory.getExecutor();
     }
 
-    protected Executor newContextExecutor(@Nullable Map<String, String> mdcContext) {
+    /**
+     * Create a new executor for the database. This is used internally when creating a transaction or a new runner.
+     * @param mdcContext if present, the MDC context to be made available within the executors threads
+     * @return the new executor
+     */
+    @API(API.Status.INTERNAL)
+    public Executor newContextExecutor(@Nullable Map<String, String> mdcContext) {
         return factory.newContextExecutor(mdcContext);
     }
 

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/runners/throttled/CursorFactory.java

@@ -0,0 +1,49 @@
+/*
+ * CursorFactory.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.apple.foundationdb.record.provider.foundationdb.runners.throttled;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.record.RecordCursor;
+import com.apple.foundationdb.record.RecordCursorResult;
+import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore;
+
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
+
+/**
+ * Create a cursor with the given store and last result.
+ * @param <T> the type of item the cursor iterates over.
+ * This factory method is used by the {@link ThrottledRetryingIterator} to create inner cursors when needed.
+ * The iterator creates transactions based off of the constraints given, and for each such transaction, a new inner
+ * cursor gets created.
+ */
+@API(API.Status.EXPERIMENTAL)
+@FunctionalInterface
+public interface CursorFactory<T> {
+    /**
+     * Create a new inner cursor for the {@link ThrottledRetryingIterator}.
+     * @param store the record store to use
+     * @param lastResult the last result processed by the previous cursor (use for continuation). Null is none.
+     * @param rowLimit the adjusted row limit to use
+     * @return a newly created cursor with the given continuation and limit
+     */
+    RecordCursor<T> createCursor(@Nonnull FDBRecordStore store, @Nullable RecordCursorResult<T> lastResult, int rowLimit);
+}

fdb-record-layer-core/src/main/java/com/apple/foundationdb/record/provider/foundationdb/runners/throttled/ItemHandler.java

@@ -0,0 +1,51 @@
+/*
+ * ItemHandler.java
+ *
+ * This source file is part of the FoundationDB open source project
+ *
+ * Copyright 2015-2025 Apple Inc. and the FoundationDB project authors
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.apple.foundationdb.record.provider.foundationdb.runners.throttled;
+
+import com.apple.foundationdb.annotation.API;
+import com.apple.foundationdb.record.RecordCursorResult;
+import com.apple.foundationdb.record.provider.foundationdb.FDBRecordStore;
+
+import javax.annotation.Nonnull;
+import java.util.concurrent.CompletableFuture;
+
+/**
+ * A handler of an item during the iteration of a {@link ThrottledRetryingIterator}.
+ * @param <T> the type of element in the iteration
+ */
+@API(API.Status.EXPERIMENTAL)
+@FunctionalInterface
+public interface ItemHandler<T> {
+    /**
+     * Process an item.
+     * Once done processing, return a future that controls whether to continue the iteration or stop.
+     * The quota manager holds the current state of the iteration (per the current transaction). The handler can
+     * change the state via {@link ThrottledRetryingIterator.QuotaManager#deleteCountAdd(int)},
+     * {@link ThrottledRetryingIterator.QuotaManager#deleteCountInc()} and
+     * {@link ThrottledRetryingIterator.QuotaManager#markExhausted()}.
+     * @param store the record store to use
+     * @param lastResult the result to process
+     * @param quotaManager the current quota manager state
+     * @return Future (Void) for when the operation is complete
+     */
+    @Nonnull
+    CompletableFuture<Void> handleOneItem(@Nonnull FDBRecordStore store, @Nonnull RecordCursorResult<T> lastResult, @Nonnull ThrottledRetryingIterator.QuotaManager quotaManager);
+}