From b5795db0cf517f8942eed868752249df9b105603 Mon Sep 17 00:00:00 2001 From: Adrien Grand Date: Tue, 7 Jun 2022 16:59:55 +0200 Subject: [PATCH] LUCENE-10078: Enable merge-on-refresh by default. (#921) This gives implementations of `findFullFlushMerges` to `LogMergePolicy` and `TieredMergePolicy` and enables merge-on-refresh with a default timeout of 500ms. The idea behind the 500ms default is that it felt both high-enough to have time to run merges of small segments, and low enough that the freshness of the data wouldn't look badly affected for users who have high refresh rates (e.g. refreshing every second). In both cases, `findFullFlushMerges` delegates to `findMerges` and filters merges whose segments are all below the min/floor size. --- lucene/CHANGES.txt | 3 ++ .../lucene/index/FilterMergePolicy.java | 5 +++ .../org/apache/lucene/index/IndexWriter.java | 8 ++++ .../lucene/index/IndexWriterConfig.java | 13 +++++-- .../lucene/index/LogByteSizeMergePolicy.java | 7 ++-- .../lucene/index/LogDocMergePolicy.java | 7 ++-- .../apache/lucene/index/LogMergePolicy.java | 13 ++++++- .../org/apache/lucene/index/MergePolicy.java | 37 +++++++++++++++++-- .../lucene/index/TieredMergePolicy.java | 22 +++++++++-- .../index/TestDirectoryReaderReopen.java | 6 ++- .../lucene/index/TestLogMergePolicy.java | 29 ++++++++++++++- .../lucene/index/TestTieredMergePolicy.java | 29 +++++++++++++++ .../index/TestTragicIndexWriterDeadlock.java | 14 ++++++- .../tests/index/BaseMergePolicyTestCase.java | 10 ++++- .../lucene/tests/util/LuceneTestCase.java | 14 +++++++ 15 files changed, 192 insertions(+), 25 deletions(-) diff --git a/lucene/CHANGES.txt b/lucene/CHANGES.txt index 1021400df74c..f3de66e7409e 100644 --- a/lucene/CHANGES.txt +++ b/lucene/CHANGES.txt @@ -76,6 +76,9 @@ New Features Improvements --------------------- +* LUCENE-10078: Merge on full flush is now enabled by default with a timeout of + 500ms. (Adrien Grand) + * LUCENE-10585: Facet module code cleanup (copy/paste scrubbing, simplification and some very minor optimization tweaks). (Greg Miller) diff --git a/lucene/core/src/java/org/apache/lucene/index/FilterMergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/FilterMergePolicy.java index 8a5e3901c405..6eeb61cde736 100644 --- a/lucene/core/src/java/org/apache/lucene/index/FilterMergePolicy.java +++ b/lucene/core/src/java/org/apache/lucene/index/FilterMergePolicy.java @@ -124,4 +124,9 @@ public int numDeletesToMerge( public MergePolicy unwrap() { return in; } + + @Override + protected long maxFullFlushMergeSize() { + return in.maxFullFlushMergeSize(); + } } diff --git a/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java b/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java index 484006fc420c..832e1724d14c 100644 --- a/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java +++ b/lucene/core/src/java/org/apache/lucene/index/IndexWriter.java @@ -2372,6 +2372,10 @@ public synchronized Set getMergingSegments() { * @lucene.experimental */ private synchronized MergePolicy.OneMerge getNextMerge() { + if (tragedy.get() != null) { + throw new IllegalStateException( + "this writer hit an unrecoverable error; cannot merge", tragedy.get()); + } if (pendingMerges.size() == 0) { return null; } else { @@ -2388,6 +2392,10 @@ private synchronized MergePolicy.OneMerge getNextMerge() { * @lucene.experimental */ public synchronized boolean hasPendingMerges() { + if (tragedy.get() != null) { + throw new IllegalStateException( + "this writer hit an unrecoverable error; cannot merge", tragedy.get()); + } return pendingMerges.size() != 0; } diff --git a/lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java b/lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java index a4ff970b58d0..c5e9a8bd2240 100644 --- a/lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java +++ b/lucene/core/src/java/org/apache/lucene/index/IndexWriterConfig.java @@ -105,7 +105,7 @@ public enum OpenMode { * Default value for time to wait for merges on commit or getReader (when using a {@link * MergePolicy} that implements {@link MergePolicy#findFullFlushMerges}). */ - public static final long DEFAULT_MAX_FULL_FLUSH_MERGE_WAIT_MILLIS = 0; + public static final long DEFAULT_MAX_FULL_FLUSH_MERGE_WAIT_MILLIS = 500; // indicates whether this config instance is already attached to a writer. // not final so that it can be cloned properly. @@ -457,9 +457,14 @@ public IndexWriterConfig setCommitOnClose(boolean commitOnClose) { * call, like natural segment merges. The default is * {@value IndexWriterConfig#DEFAULT_MAX_FULL_FLUSH_MERGE_WAIT_MILLIS}. * - *

Note: This settings has no effect unless {@link - * MergePolicy#findFullFlushMerges(MergeTrigger, SegmentInfos, MergePolicy.MergeContext)} has an - * implementation that actually returns merges which by default doesn't return any merges. + *

Note: Which segments would get merged depends on the implementation of {@link + * MergePolicy#findFullFlushMerges(MergeTrigger, SegmentInfos, MergePolicy.MergeContext)} + * + *

Note: Set to 0 to disable merging on full flush. + * + *

Note: If {@link SerialMergeScheduler} is used and a non-zero timout is configured, + * full-flush merges will always wait for the merge to finish without honoring the configured + * timeout. */ public IndexWriterConfig setMaxFullFlushMergeWaitMillis(long maxFullFlushMergeWaitMillis) { this.maxFullFlushMergeWaitMillis = maxFullFlushMergeWaitMillis; diff --git a/lucene/core/src/java/org/apache/lucene/index/LogByteSizeMergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/LogByteSizeMergePolicy.java index 6f0aa5cde484..f858d1d8f3b4 100644 --- a/lucene/core/src/java/org/apache/lucene/index/LogByteSizeMergePolicy.java +++ b/lucene/core/src/java/org/apache/lucene/index/LogByteSizeMergePolicy.java @@ -96,9 +96,10 @@ public double getMaxMergeMBForForcedMerge() { } /** - * Sets the minimum size for the lowest level segments. Any segments below this size will be - * merged more aggressively in order to avoid having a long tail of small segments. Large values - * of this parameter increase the merging cost during indexing if you flush small segments. + * Sets the minimum size for the lowest level segments. Any segments below this size are + * candidates for full-flush merges and be merged more aggressively in order to avoid having a + * long tail of small segments. Large values of this parameter increase the merging cost during + * indexing if you flush small segments. */ public void setMinMergeMB(double mb) { minMergeSize = (long) (mb * 1024 * 1024); diff --git a/lucene/core/src/java/org/apache/lucene/index/LogDocMergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/LogDocMergePolicy.java index 5f26d91ca556..6f9da9edc482 100644 --- a/lucene/core/src/java/org/apache/lucene/index/LogDocMergePolicy.java +++ b/lucene/core/src/java/org/apache/lucene/index/LogDocMergePolicy.java @@ -43,9 +43,10 @@ protected long size(SegmentCommitInfo info, MergeContext mergeContext) throws IO } /** - * Sets the minimum size for the lowest level segments. Any segments below this size will be - * merged more aggressively in order to avoid having a long tail of small segments. Large values - * of this parameter increase the merging cost during indexing if you flush small segments. + * Sets the minimum size for the lowest level segments. Any segments below this size are + * candidates for full-flush merges and merged more aggressively in order to avoid having a long + * tail of small segments. Large values of this parameter increase the merging cost during + * indexing if you flush small segments. */ public void setMinMergeDocs(int minMergeDocs) { minMergeSize = minMergeDocs; diff --git a/lucene/core/src/java/org/apache/lucene/index/LogMergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/LogMergePolicy.java index c279bf34832f..1d80ba6dfd6c 100644 --- a/lucene/core/src/java/org/apache/lucene/index/LogMergePolicy.java +++ b/lucene/core/src/java/org/apache/lucene/index/LogMergePolicy.java @@ -35,6 +35,9 @@ * specifies how a segment's size is determined. {@link LogDocMergePolicy} is one subclass that * measures size by document count in the segment. {@link LogByteSizeMergePolicy} is another * subclass that measures size as the total byte size of the file(s) for the segment. + * + *

NOTE: This policy returns natural merges whose size is below the {@link #minMergeSize + * minimum merge size} for {@link #findFullFlushMerges full-flush merges}. */ public abstract class LogMergePolicy extends MergePolicy { @@ -64,7 +67,10 @@ public abstract class LogMergePolicy extends MergePolicy { /** How many segments to merge at a time. */ protected int mergeFactor = DEFAULT_MERGE_FACTOR; - /** Any segments whose size is smaller than this value will be merged more aggressively. */ + /** + * Any segments whose size is smaller than this value will be candidates for full-flush merges and + * merged more aggressively. + */ protected long minMergeSize; /** If the size of a segment exceeds this value then it will never be merged. */ @@ -178,6 +184,11 @@ protected boolean isMerged( && (numToMerge != 1 || !segmentIsOriginal || isMerged(infos, mergeInfo, mergeContext)); } + @Override + protected long maxFullFlushMergeSize() { + return minMergeSize; + } + /** * Returns the merges necessary to merge the index, taking the max merge size or max merge docs * into consideration. This method attempts to respect the {@code maxNumSegments} parameter, diff --git a/lucene/core/src/java/org/apache/lucene/index/MergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/MergePolicy.java index c3e99b817be9..b84958ab6092 100644 --- a/lucene/core/src/java/org/apache/lucene/index/MergePolicy.java +++ b/lucene/core/src/java/org/apache/lucene/index/MergePolicy.java @@ -601,9 +601,9 @@ public abstract MergeSpecification findForcedDeletesMerges( SegmentInfos segmentInfos, MergeContext mergeContext) throws IOException; /** - * Identifies merges that we want to execute (synchronously) on commit. By default, this will do - * no merging on commit. If you implement this method in your {@code MergePolicy} you must also - * set a non-zero timeout using {@link IndexWriterConfig#setMaxFullFlushMergeWaitMillis}. + * Identifies merges that we want to execute (synchronously) on commit. By default, this will + * return {@link #findMerges natural merges} whose segments are all less than the {@link + * #maxFullFlushMergeSize() max segment size for full flushes}. * *

Any merges returned here will make {@link IndexWriter#commit()}, {@link * IndexWriter#prepareCommit()} or {@link IndexWriter#getReader(boolean, boolean)} block until the @@ -628,7 +628,28 @@ public abstract MergeSpecification findForcedDeletesMerges( public MergeSpecification findFullFlushMerges( MergeTrigger mergeTrigger, SegmentInfos segmentInfos, MergeContext mergeContext) throws IOException { - return null; + // This returns natural merges that contain segments below the minimum size + MergeSpecification mergeSpec = findMerges(mergeTrigger, segmentInfos, mergeContext); + if (mergeSpec == null) { + return null; + } + MergeSpecification newMergeSpec = null; + for (OneMerge oneMerge : mergeSpec.merges) { + boolean belowMaxFullFlushSize = true; + for (SegmentCommitInfo sci : oneMerge.segments) { + if (size(sci, mergeContext) >= maxFullFlushMergeSize()) { + belowMaxFullFlushSize = false; + break; + } + } + if (belowMaxFullFlushSize) { + if (newMergeSpec == null) { + newMergeSpec = new MergeSpecification(); + } + newMergeSpec.add(oneMerge); + } + } + return newMergeSpec; } /** @@ -671,6 +692,14 @@ protected long size(SegmentCommitInfo info, MergeContext mergeContext) throws IO return (info.info.maxDoc() <= 0 ? byteSize : (long) (byteSize * (1.0 - delRatio))); } + /** + * Return the maximum size of segments to be included in full-flush merges by the default + * implementation of {@link #findFullFlushMerges}. + */ + protected long maxFullFlushMergeSize() { + return 0L; + } + /** Asserts that the delCount for this SegmentCommitInfo is valid */ protected final boolean assertDelCount(int delCount, SegmentCommitInfo info) { assert delCount >= 0 : "delCount must be positive: " + delCount; diff --git a/lucene/core/src/java/org/apache/lucene/index/TieredMergePolicy.java b/lucene/core/src/java/org/apache/lucene/index/TieredMergePolicy.java index 0a98760b7ef9..de394365dbe3 100644 --- a/lucene/core/src/java/org/apache/lucene/index/TieredMergePolicy.java +++ b/lucene/core/src/java/org/apache/lucene/index/TieredMergePolicy.java @@ -62,6 +62,10 @@ * *

findForcedDeletesMerges should never produce segments greater than maxSegmentSize. * + *

NOTE: This policy returns natural merges whose size is below the {@link + * #setFloorSegmentMB(double) floor segment size} for {@link #findFullFlushMerges full-flush + * merges}. + * * @lucene.experimental */ @@ -168,9 +172,16 @@ public double getDeletesPctAllowed() { } /** - * Segments smaller than this are "rounded up" to this size, ie treated as equal (floor) size for - * merge selection. This is to prevent frequent flushing of tiny segments from allowing a long - * tail in the index. Default is 2 MB. + * Segments smaller than this size are merged more aggressively: + * + *

+ * + * In both cases, this helps prevent frequent flushing of tiny segments to create a long tail of + * small segments in the index. Default is 2MB. */ public TieredMergePolicy setFloorSegmentMB(double v) { if (v <= 0.0) { @@ -190,6 +201,11 @@ public double getFloorSegmentMB() { return floorSegmentBytes / (1024 * 1024.); } + @Override + protected long maxFullFlushMergeSize() { + return floorSegmentBytes; + } + /** * When forceMergeDeletes is called, we only merge away a segment if its delete percentage is over * this threshold. Default is 10%. diff --git a/lucene/core/src/test/org/apache/lucene/index/TestDirectoryReaderReopen.java b/lucene/core/src/test/org/apache/lucene/index/TestDirectoryReaderReopen.java index 462343f6bcf7..4dd322b09c92 100644 --- a/lucene/core/src/test/org/apache/lucene/index/TestDirectoryReaderReopen.java +++ b/lucene/core/src/test/org/apache/lucene/index/TestDirectoryReaderReopen.java @@ -815,7 +815,8 @@ public void testNPEAfterInvalidReindex2() throws Exception { /** test reopening backwards from a non-NRT reader (with document deletes) */ public void testNRTMdeletes() throws Exception { Directory dir = newDirectory(); - IndexWriterConfig iwc = new IndexWriterConfig(new MockAnalyzer(random())); + IndexWriterConfig iwc = + new IndexWriterConfig(new MockAnalyzer(random())).setMergePolicy(NoMergePolicy.INSTANCE); SnapshotDeletionPolicy snapshotter = new SnapshotDeletionPolicy(new KeepOnlyLastCommitDeletionPolicy()); iwc.setIndexDeletionPolicy(snapshotter); @@ -865,7 +866,8 @@ public void testNRTMdeletes() throws Exception { /** test reopening backwards from an NRT reader (with document deletes) */ public void testNRTMdeletes2() throws Exception { Directory dir = newDirectory(); - IndexWriterConfig iwc = new IndexWriterConfig(new MockAnalyzer(random())); + IndexWriterConfig iwc = + new IndexWriterConfig(new MockAnalyzer(random())).setMergePolicy(NoMergePolicy.INSTANCE); SnapshotDeletionPolicy snapshotter = new SnapshotDeletionPolicy(new KeepOnlyLastCommitDeletionPolicy()); iwc.setIndexDeletionPolicy(snapshotter); diff --git a/lucene/core/src/test/org/apache/lucene/index/TestLogMergePolicy.java b/lucene/core/src/test/org/apache/lucene/index/TestLogMergePolicy.java index 517542699c14..dce1e5acb17e 100644 --- a/lucene/core/src/test/org/apache/lucene/index/TestLogMergePolicy.java +++ b/lucene/core/src/test/org/apache/lucene/index/TestLogMergePolicy.java @@ -27,7 +27,7 @@ public class TestLogMergePolicy extends BaseMergePolicyTestCase { @Override - public MergePolicy mergePolicy() { + public LogMergePolicy mergePolicy() { return newLogMergePolicy(random()); } @@ -187,4 +187,31 @@ public void testRejectUnbalancedMerges() throws IOException { assertEquals(100, segmentInfos.info(0).info.maxDoc()); assertEquals(10, segmentInfos.info(1).info.maxDoc()); } + + public void testFullFlushMerges() throws IOException { + AtomicLong segNameGenerator = new AtomicLong(); + IOStats stats = new IOStats(); + MergeContext mergeContext = new MockMergeContext(SegmentCommitInfo::getDelCount); + SegmentInfos segmentInfos = new SegmentInfos(Version.LATEST.major); + + LogMergePolicy mp = mergePolicy(); + + for (int i = 0; i < mp.getMergeFactor(); ++i) { + segmentInfos.add( + makeSegmentCommitInfo( + "_" + segNameGenerator.getAndIncrement(), + 1, + 0, + Double.MIN_VALUE, + IndexWriter.SOURCE_FLUSH)); + } + MergeSpecification spec = + mp.findFullFlushMerges(MergeTrigger.FULL_FLUSH, segmentInfos, mergeContext); + assertNotNull(spec); + for (OneMerge merge : spec.merges) { + segmentInfos = + applyMerge(segmentInfos, merge, "_" + segNameGenerator.getAndIncrement(), stats); + } + assertEquals(1, segmentInfos.size()); + } } diff --git a/lucene/core/src/test/org/apache/lucene/index/TestTieredMergePolicy.java b/lucene/core/src/test/org/apache/lucene/index/TestTieredMergePolicy.java index cfeff53f888b..0523d9db21bb 100644 --- a/lucene/core/src/test/org/apache/lucene/index/TestTieredMergePolicy.java +++ b/lucene/core/src/test/org/apache/lucene/index/TestTieredMergePolicy.java @@ -24,9 +24,11 @@ import java.util.List; import java.util.Locale; import java.util.Map; +import java.util.concurrent.atomic.AtomicLong; import org.apache.lucene.document.Document; import org.apache.lucene.document.Field; import org.apache.lucene.document.StoredField; +import org.apache.lucene.index.MergePolicy.MergeContext; import org.apache.lucene.index.MergePolicy.MergeSpecification; import org.apache.lucene.index.MergePolicy.OneMerge; import org.apache.lucene.store.Directory; @@ -917,4 +919,31 @@ public void testSimulateUpdates() throws IOException { int numDocs = TEST_NIGHTLY ? atLeast(10_000_000) : atLeast(1_000_000); doTestSimulateUpdates(mergePolicy, numDocs, 2500); } + + public void testFullFlushMerges() throws IOException { + AtomicLong segNameGenerator = new AtomicLong(); + IOStats stats = new IOStats(); + MergeContext mergeContext = new MockMergeContext(SegmentCommitInfo::getDelCount); + SegmentInfos segmentInfos = new SegmentInfos(Version.LATEST.major); + + TieredMergePolicy mp = new TieredMergePolicy(); + + for (int i = 0; i < 11; ++i) { + segmentInfos.add( + makeSegmentCommitInfo( + "_" + segNameGenerator.getAndIncrement(), + 1, + 0, + Double.MIN_VALUE, + IndexWriter.SOURCE_FLUSH)); + } + MergeSpecification spec = + mp.findFullFlushMerges(MergeTrigger.FULL_FLUSH, segmentInfos, mergeContext); + assertNotNull(spec); + for (OneMerge merge : spec.merges) { + segmentInfos = + applyMerge(segmentInfos, merge, "_" + segNameGenerator.getAndIncrement(), stats); + } + assertEquals(2, segmentInfos.size()); + } } diff --git a/lucene/core/src/test/org/apache/lucene/index/TestTragicIndexWriterDeadlock.java b/lucene/core/src/test/org/apache/lucene/index/TestTragicIndexWriterDeadlock.java index 8abbfca8d218..37f176ea030e 100644 --- a/lucene/core/src/test/org/apache/lucene/index/TestTragicIndexWriterDeadlock.java +++ b/lucene/core/src/test/org/apache/lucene/index/TestTragicIndexWriterDeadlock.java @@ -104,8 +104,17 @@ public void run() { // LUCENE-7570 public void testDeadlockStalledMerges() throws Exception { + doTestDeadlockStalledMerges(false); + } + + public void testDeadlockStalledFullFlushMerges() throws Exception { + doTestDeadlockStalledMerges(true); + } + + private void doTestDeadlockStalledMerges(boolean mergeOnFlush) throws Exception { Directory dir = newDirectory(); - IndexWriterConfig iwc = new IndexWriterConfig(); + IndexWriterConfig iwc = + new IndexWriterConfig().setMaxFullFlushMergeWaitMillis(mergeOnFlush ? 1000 : 0); // so we merge every 2 segments: LogMergePolicy mp = new LogDocMergePolicy(); @@ -163,7 +172,8 @@ protected void mergeSuccess(MergePolicy.OneMerge merge) { w.addDocument(new Document()); // w writes third segment w.addDocument(new Document()); - w.commit(); + IllegalStateException e = expectThrows(IllegalStateException.class, () -> w.commit()); + assertTrue(e.getMessage(), e.getMessage().startsWith("this writer hit an unrecoverable error")); // w writes fourth segment, and commit flushes and kicks off merge that stalls w.close(); dir.close(); diff --git a/lucene/test-framework/src/java/org/apache/lucene/tests/index/BaseMergePolicyTestCase.java b/lucene/test-framework/src/java/org/apache/lucene/tests/index/BaseMergePolicyTestCase.java index 4dcf5e0cf0ae..8dd618afd4aa 100644 --- a/lucene/test-framework/src/java/org/apache/lucene/tests/index/BaseMergePolicyTestCase.java +++ b/lucene/test-framework/src/java/org/apache/lucene/tests/index/BaseMergePolicyTestCase.java @@ -428,7 +428,10 @@ protected void doTestSimulateAppendOnly( IndexWriter.SOURCE_FLUSH)); MergeSpecification merges = - mergePolicy.findMerges(MergeTrigger.SEGMENT_FLUSH, segmentInfos, mergeContext); + mergePolicy.findFullFlushMerges(MergeTrigger.SEGMENT_FLUSH, segmentInfos, mergeContext); + if (merges == null) { + merges = mergePolicy.findMerges(MergeTrigger.SEGMENT_FLUSH, segmentInfos, mergeContext); + } while (merges != null) { assertTrue(merges.merges.size() > 0); assertMerge(mergePolicy, merges); @@ -490,7 +493,10 @@ protected void doTestSimulateUpdates(MergePolicy mergePolicy, int totalDocs, int flushSize, IndexWriter.SOURCE_FLUSH)); MergeSpecification merges = - mergePolicy.findMerges(MergeTrigger.SEGMENT_FLUSH, segmentInfos, mergeContext); + mergePolicy.findFullFlushMerges(MergeTrigger.SEGMENT_FLUSH, segmentInfos, mergeContext); + if (merges == null) { + merges = mergePolicy.findMerges(MergeTrigger.SEGMENT_FLUSH, segmentInfos, mergeContext); + } while (merges != null) { assertMerge(mergePolicy, merges); for (OneMerge oneMerge : merges.merges) { diff --git a/lucene/test-framework/src/java/org/apache/lucene/tests/util/LuceneTestCase.java b/lucene/test-framework/src/java/org/apache/lucene/tests/util/LuceneTestCase.java index 43af8c6769c4..7777424ff692 100644 --- a/lucene/test-framework/src/java/org/apache/lucene/tests/util/LuceneTestCase.java +++ b/lucene/test-framework/src/java/org/apache/lucene/tests/util/LuceneTestCase.java @@ -1041,6 +1041,20 @@ protected synchronized boolean maybeStall(MergeSource mergeSource) { if (rarely(r)) { c.setIndexWriterEventListener(new MockIndexWriterEventListener()); } + switch (r.nextInt(3)) { + case 0: + // Disable merge on refresh + c.setMaxFullFlushMergeWaitMillis(0L); + break; + case 1: + // Very low timeout, merges will likely not be able to run in time + c.setMaxFullFlushMergeWaitMillis(1L); + break; + default: + // Very long timeout, merges will almost always be able to run in time + c.setMaxFullFlushMergeWaitMillis(1000L); + break; + } c.setMaxFullFlushMergeWaitMillis(rarely() ? atLeast(r, 1000) : atLeast(r, 200)); return c;