wip

Author: jjezra

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

@@ -186,6 +186,19 @@ public static Builder newBuilder() {
         return new Builder();
     }
 
+
+    /**
+     * Not used anymore.
+     * @return always false;
+     * @deprecated see {@link Builder#setUseSynchronizedSession(boolean)}
+     */
+    @API(API.Status.DEPRECATED)
+    @SuppressWarnings("PMD.AvoidUsingHardCodedIP") // version is not IP
+    @Deprecated(since = "4.4.3.0", forRemoval = true)
+    public boolean shouldUseSynchronizedSession() {
+        return false;
+    }
+
     public long getLeaseLengthMillis() {
         return leaseLengthMillis;
     }
@@ -496,8 +509,9 @@ public Builder setUseSynchronizedSession(boolean useSynchronizedSession) {
         }
 
         /**
-         * Set the lease length in milliseconds if the synchronized session is used. By default this is {@link #DEFAULT_LEASE_LENGTH_MILLIS}.
-         * @see com.apple.foundationdb.synchronizedsession.SynchronizedSession
+         * If the indexing session is not expected to be mutual, abort indexing if another session is active. This function
+         * defines the maximum age of another session's heartbeat to be considered an "active session".
+         * The default value is {@link #DEFAULT_LEASE_LENGTH_MILLIS}.
          * @param leaseLengthMillis length between last access and lease's end time in milliseconds
          * @return this builder
          */

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

@@ -29,6 +29,7 @@
 import com.apple.foundationdb.record.metadata.expressions.EmptyKeyExpression;
 import com.apple.foundationdb.record.metadata.expressions.GroupingKeyExpression;
 import com.apple.foundationdb.record.metadata.expressions.KeyExpression;
+import com.apple.foundationdb.synchronizedsession.SynchronizedSessionLockedException;
 import com.apple.test.BooleanSource;
 import com.google.common.collect.Comparators;
 import org.junit.jupiter.api.Test;
@@ -40,6 +41,9 @@
 import javax.annotation.Nullable;
 import java.util.List;
 import java.util.Map;
+import java.util.concurrent.Semaphore;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicBoolean;
 import java.util.concurrent.atomic.AtomicLong;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
@@ -1156,4 +1160,69 @@ void testIndexFromIndexBlock() {
         assertReadable(tgtIndex);
         scrubAndValidate(List.of(tgtIndex));
     }
+
+    @Test
+    void testForbidConcurrentIndexFromIndexSessions() throws InterruptedException {
+        // Do not let a conversion of few indexes of an active multi-target session
+        final int numRecords = 59;
+        populateData(numRecords);
+
+        Index sourceIndex = new Index("src_index", field("num_value_2"), EmptyKeyExpression.EMPTY, IndexTypes.VALUE, IndexOptions.UNIQUE_OPTIONS);
+        openSimpleMetaData(metaDataBuilder -> metaDataBuilder.addIndex("MySimpleRecord", sourceIndex));
+        buildIndexClean(sourceIndex);
+
+        // Partly build index
+        Index tgtIndex = new Index("tgt_index", field("num_value_3_indexed"), IndexTypes.VALUE);
+        FDBRecordStoreTestBase.RecordMetaDataHook hook = myHook(sourceIndex, tgtIndex);
+        openSimpleMetaData(hook);
+
+        Semaphore pauseMutualBuildSemaphore = new Semaphore(1);
+        Semaphore startBuildingSemaphore =  new Semaphore(1);
+        pauseMutualBuildSemaphore.acquire();
+        startBuildingSemaphore.acquire();
+        AtomicBoolean passed = new AtomicBoolean(false);
+        Thread t1 = new Thread(() -> {
+            // build index and pause halfway, allowing an active session test
+            try (OnlineIndexer indexBuilder = newIndexerBuilder(tgtIndex)
+                    .setLeaseLengthMillis(TimeUnit.SECONDS.toMillis(20))
+                    .setLimit(4)
+                    .setIndexingPolicy(OnlineIndexer.IndexingPolicy.newBuilder()
+                            .setSourceIndex("src_index")
+                            .forbidRecordScan())
+                    .setConfigLoader(old -> {
+                        if (passed.get()) {
+                            try {
+                                startBuildingSemaphore.release();
+                                pauseMutualBuildSemaphore.acquire(); // pause to try building indexes
+                            } catch (InterruptedException e) {
+                                throw new RuntimeException(e);
+                            } finally {
+                                pauseMutualBuildSemaphore.release();
+                            }
+                        } else {
+                            passed.set(true);
+                        }
+                        return old;
+                    })
+                    .build()) {
+                indexBuilder.buildIndex();
+            }
+        });
+        t1.start();
+        startBuildingSemaphore.acquire();
+        startBuildingSemaphore.release();
+        // Try one index at a time
+        try (OnlineIndexer indexBuilder = newIndexerBuilder(tgtIndex)
+                .setIndexingPolicy(OnlineIndexer.IndexingPolicy.newBuilder()
+                        .setSourceIndex("src_index")
+                        .forbidRecordScan())
+                .build()) {
+            assertThrows(SynchronizedSessionLockedException.class, indexBuilder::buildIndex);
+        }
+        // let the other thread finish indexing
+        pauseMutualBuildSemaphore.release();
+        t1.join();
+        // happy indexes assertion
+        assertReadable(tgtIndex);
+    }
 }

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

@@ -887,4 +887,32 @@ public void runWithWeakReadSemantics() {
             fdb.setTrackLastSeenVersionOnRead(dbTracksReadVersionOnCommit);
         }
     }
+
+    @Test
+    @SuppressWarnings("removal")
+    void testDeprecatedSetUseSynchronizedSession() {
+        List<TestRecords1Proto.MySimpleRecord> records = LongStream.range(0, 20).mapToObj(val ->
+                TestRecords1Proto.MySimpleRecord.newBuilder().setRecNo(val).setNumValue2((int)val + 1).build()
+        ).collect(Collectors.toList());
+        Index index = new Index("simple$value_2", field("num_value_2").ungrouped(), IndexTypes.SUM);
+        FDBRecordStoreTestBase.RecordMetaDataHook hook = metaDataBuilder -> metaDataBuilder.addIndex("MySimpleRecord", index);
+
+        openSimpleMetaData();
+        try (FDBRecordContext context = openContext()) {
+            records.forEach(recordStore::saveRecord);
+            context.commit();
+        }
+
+        openSimpleMetaData(hook);
+        disableAll(List.of(index));
+
+        openSimpleMetaData(hook);
+        try (OnlineIndexer indexBuilder = newIndexerBuilder(index)
+                .setUseSynchronizedSession(true)
+                .build()) {
+            indexBuilder.buildIndex();
+        }
+
+        assertReadable(index);
+    }
 }

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

@@ -76,7 +76,7 @@ void testHeartbeatLowLevel() {
             Map<UUID, IndexBuildProto.IndexBuildHeartbeat> queried = indexer.getIndexingHeartbeats(0);
             Assertions.assertThat(queried).hasSize(count);
             Assertions.assertThat(queried.keySet())
-                    .containsExactlyInAnyOrderElementsOf(Arrays.stream(heartbeats).map(ht -> ht.indexerId).collect(Collectors.toList()));
+                    .containsExactlyInAnyOrderElementsOf(Arrays.stream(heartbeats).map(heartbeat -> heartbeat.indexerId).collect(Collectors.toList()));
 
             // Query, partial
             queried = indexer.getIndexingHeartbeats(5);