This is an automated email from the ASF dual-hosted git repository.
adulceanu pushed a commit to branch issues/OAK-9922
in repository https://gitbox.apache.org/repos/asf/jackrabbit-oak.git
commit 9402ecb9fe028ad615f7a8a965d17f5b267d4888
Author: Lucas Weitzendorf <lweitzend...@adobe.com>
AuthorDate: Tue Oct 11 16:32:42 2022 +0200
OAK-9922 Parallel Compaction
Make SegmentBufferWriterPool implementation a configurable option
---
.../oak/segment/DefaultSegmentWriterBuilder.java | 60 ++--
.../oak/segment/SegmentBufferWriterPool.java | 314 +++++++++++++++++----
.../jackrabbit/oak/segment/file/FileStore.java | 12 +-
.../jackrabbit/oak/segment/memory/MemoryStore.java | 16 +-
.../jackrabbit/oak/segment/NodeRecordTest.java | 2 +-
.../segment/ParallelCompactorExternalBlobTest.java | 2 +-
.../oak/segment/ParallelCompactorTest.java | 2 +-
.../oak/segment/SegmentBufferWriterPoolTest.java | 25 +-
8 files changed, 310 insertions(+), 123 deletions(-)
diff --git
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DefaultSegmentWriterBuilder.java
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DefaultSegmentWriterBuilder.java
index 94ae636a4b..4d70c23c2c 100644
---
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DefaultSegmentWriterBuilder.java
+++
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/DefaultSegmentWriterBuilder.java
@@ -20,6 +20,7 @@
package org.apache.jackrabbit.oak.segment;
import static
org.apache.jackrabbit.guava.common.base.Preconditions.checkNotNull;
+import static
org.apache.jackrabbit.oak.segment.SegmentBufferWriterPool.PoolType;
import org.apache.jackrabbit.guava.common.base.Supplier;
import org.apache.jackrabbit.guava.common.base.Suppliers;
@@ -29,10 +30,11 @@ import
org.apache.jackrabbit.oak.segment.file.ReadOnlyFileStore;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
import org.jetbrains.annotations.NotNull;
+import org.jetbrains.annotations.Nullable;
/**
* Builder for building {@link DefaultSegmentWriter} instances.
- * The returned instances are thread safe if {@link #withWriterPool()}
+ * The returned instances are thread safe if {@link #withWriterPool(PoolType)}
* was specified and <em>not</em> thread sage if {@link #withoutWriterPool()}
* was specified (default).
* <p>
@@ -56,7 +58,7 @@ public final class DefaultSegmentWriterBuilder {
@NotNull
private Supplier<GCGeneration> generation =
Suppliers.ofInstance(GCGeneration.NULL);
- private boolean pooled = false;
+ private PoolType poolType = null;
@NotNull
private WriterCacheManager cacheManager = new WriterCacheManager.Default();
@@ -81,7 +83,7 @@ public final class DefaultSegmentWriterBuilder {
* If {@link #withoutWriterPool()} was specified all segments will be
written
* at the generation that {@code generation.get()} returned at the time
* any of the {@code build()} methods is called.
- * If {@link #withWriterPool()} was specified a segments will be written
+ * If {@link #withWriterPool(PoolType)} ()} was specified, segments will
be written
* at the generation that {@code generation.get()} returns when a new
segment
* is created by the returned writer.
*/
@@ -106,8 +108,8 @@ public final class DefaultSegmentWriterBuilder {
* The returned instance is thread safe.
*/
@NotNull
- public DefaultSegmentWriterBuilder withWriterPool() {
- this.pooled = true;
+ public DefaultSegmentWriterBuilder withWriterPool(PoolType writerType) {
+ this.poolType = writerType;
return this;
}
@@ -117,7 +119,7 @@ public final class DefaultSegmentWriterBuilder {
*/
@NotNull
public DefaultSegmentWriterBuilder withoutWriterPool() {
- this.pooled = false;
+ this.poolType = null;
return this;
}
@@ -151,7 +153,7 @@ public final class DefaultSegmentWriterBuilder {
store.getSegmentIdProvider(),
store.getBlobStore(),
cacheManager,
- createWriter(store, pooled),
+ createWriter(store, poolType),
store.getBinariesInlineThreshold()
);
}
@@ -204,47 +206,27 @@ public final class DefaultSegmentWriterBuilder {
store.getSegmentIdProvider(),
store.getBlobStore(),
cacheManager,
- createWriter(store, pooled),
+ createWriter(store, poolType),
Segment.MEDIUM_LIMIT
);
}
@NotNull
- private WriteOperationHandler createWriter(@NotNull FileStore store,
boolean pooled) {
- if (pooled) {
- return new SegmentBufferWriterPool(
- store.getSegmentIdProvider(),
- store.getReader(),
- name,
- generation
- );
- } else {
- return new SegmentBufferWriter(
- store.getSegmentIdProvider(),
- store.getReader(),
- name,
- generation.get()
- );
- }
+ private WriteOperationHandler createWriter(@NotNull FileStore store,
@Nullable PoolType poolType) {
+ return createWriter(store.getSegmentIdProvider(), store.getReader(),
poolType);
+ }
+
+ @NotNull
+ private WriteOperationHandler createWriter(@NotNull MemoryStore store,
@Nullable PoolType poolType) {
+ return createWriter(store.getSegmentIdProvider(), store.getReader(),
poolType);
}
@NotNull
- private WriteOperationHandler createWriter(@NotNull MemoryStore store,
boolean pooled) {
- if (pooled) {
- return new SegmentBufferWriterPool(
- store.getSegmentIdProvider(),
- store.getReader(),
- name,
- generation
- );
+ private WriteOperationHandler createWriter(@NotNull SegmentIdProvider
idProvider, @NotNull SegmentReader reader, @Nullable PoolType poolType) {
+ if (poolType == null) {
+ return new SegmentBufferWriter(idProvider, reader, name,
generation.get());
} else {
- return new SegmentBufferWriter(
- store.getSegmentIdProvider(),
- store.getReader(),
- name,
- generation.get()
- );
+ return SegmentBufferWriterPool.factory(idProvider, reader, name,
generation).newPool(poolType);
}
}
-
}
diff --git
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java
index b37ec2ad01..1ee0361744 100644
---
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java
+++
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPool.java
@@ -29,6 +29,9 @@ import static java.util.Objects.requireNonNull;
import java.io.IOException;
import java.util.AbstractMap.SimpleImmutableEntry;
+import java.util.List;
+import java.util.Map;
+import java.util.Set;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;
@@ -45,20 +48,7 @@ import org.jetbrains.annotations.NotNull;
* <p>
* Instances of this class are thread safe.
*/
-public class SegmentBufferWriterPool implements WriteOperationHandler {
- /**
- * Read write lock protecting the state of this pool. Multiple threads can
access their writers in parallel,
- * acquiring the read lock. The writer lock is needed for the flush
operation since it requires none
- * of the writers to be in use.
- */
- private final ReadWriteLock lock = new ReentrantReadWriteLock(true);
-
- /**
- * Pool of writers. Every thread is assigned a unique writer per GC
generation, therefore only requiring
- * a concurrent map to synchronize access to them.
- */
- private final ConcurrentMap<Object, SegmentBufferWriter> writers =
newConcurrentMap();
-
+public abstract class SegmentBufferWriterPool implements WriteOperationHandler
{
@NotNull
private final SegmentIdProvider idProvider;
@@ -73,7 +63,7 @@ public class SegmentBufferWriterPool implements
WriteOperationHandler {
private short writerId = -1;
- public SegmentBufferWriterPool(
+ private SegmentBufferWriterPool(
@NotNull SegmentIdProvider idProvider,
@NotNull SegmentReader reader,
@NotNull String wid,
@@ -84,55 +74,277 @@ public class SegmentBufferWriterPool implements
WriteOperationHandler {
this.gcGeneration = requireNonNull(gcGeneration);
}
- @NotNull
- @Override
- public GCGeneration getGCGeneration() {
- return gcGeneration.get();
+ public enum PoolType {
+ GLOBAL,
+ THREAD_SPECIFIC;
}
- @Override
- public void flush(@NotNull SegmentStore store) throws IOException {
- lock.writeLock().lock();
- try {
- for (SegmentBufferWriter writer : writers.values()) {
+ public static class SegmentBufferWriterPoolFactory {
+ @NotNull
+ private final SegmentIdProvider idProvider;
+ @NotNull
+ private final SegmentReader reader;
+ @NotNull
+ private final String wid;
+ @NotNull
+ private final Supplier<GCGeneration> gcGeneration;
+
+ private SegmentBufferWriterPoolFactory(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ this.idProvider = requireNonNull(idProvider);
+ this.reader = requireNonNull(reader);
+ this.wid = requireNonNull(wid);
+ this.gcGeneration = requireNonNull(gcGeneration);
+ }
+
+ @NotNull
+ public SegmentBufferWriterPool newPool(@NotNull
SegmentBufferWriterPool.PoolType poolType) {
+ switch (poolType) {
+ case GLOBAL:
+ return new GlobalSegmentBufferWriterPool(idProvider,
reader, wid, gcGeneration);
+ case THREAD_SPECIFIC:
+ return new
ThreadSpecificSegmentBufferWriterPool(idProvider, reader, wid, gcGeneration);
+ default:
+ throw new IllegalArgumentException("Unknown writer pool
type.");
+ }
+ }
+ }
+
+ public static SegmentBufferWriterPoolFactory factory(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ return new SegmentBufferWriterPoolFactory(idProvider, reader, wid,
gcGeneration);
+ }
+
+ private static class ThreadSpecificSegmentBufferWriterPool extends
SegmentBufferWriterPool {
+ /**
+ * Read write lock protecting the state of this pool. Multiple threads
can access their writers in parallel,
+ * acquiring the read lock. The writer lock is needed for the flush
operation since it requires none
+ * of the writers to be in use.
+ */
+ private final ReadWriteLock lock = new ReentrantReadWriteLock(true);
+
+ /**
+ * Pool of writers. Every thread is assigned a unique writer per GC
generation, therefore only requiring
+ * a concurrent map to synchronize access to them.
+ */
+ private final ConcurrentMap<Object, SegmentBufferWriter> writers =
newConcurrentMap();
+
+ public ThreadSpecificSegmentBufferWriterPool(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ super(idProvider, reader, wid, gcGeneration);
+ }
+
+ @NotNull
+ @Override
+ public RecordId execute(@NotNull GCGeneration gcGeneration,
+ @NotNull WriteOperation writeOperation)
+ throws IOException {
+ lock.readLock().lock();
+ SegmentBufferWriter writer = getWriter(currentThread(),
gcGeneration);
+ try {
+ return writeOperation.execute(writer);
+ } finally {
+ lock.readLock().unlock();
+ }
+ }
+
+ @Override
+ public void flush(@NotNull SegmentStore store) throws IOException {
+ lock.writeLock().lock();
+ try {
+ for (SegmentBufferWriter writer : writers.values()) {
+ writer.flush(store);
+ }
+ writers.clear();
+ } finally {
+ lock.writeLock().unlock();
+ }
+ }
+
+ private SegmentBufferWriter getWriter(@NotNull Thread thread, @NotNull
GCGeneration gcGeneration) {
+ SimpleImmutableEntry<?,?> key = new SimpleImmutableEntry<>(thread,
gcGeneration);
+ SegmentBufferWriter writer = writers.get(key);
+ if (writer == null) {
+ writer = newWriter(gcGeneration);
+ writers.put(key, writer);
+ }
+ return writer;
+ }
+ }
+
+ private static class GlobalSegmentBufferWriterPool extends
SegmentBufferWriterPool {
+ /**
+ * Monitor protecting the state of this pool. Neither of {@link
#writers},
+ * {@link #borrowed} and {@link #disposed} must be modified without
owning
+ * this monitor.
+ */
+ private final Monitor poolMonitor = new Monitor(true);
+
+ /**
+ * Pool of current writers that are not in use
+ */
+ private final Map<Object, SegmentBufferWriter> writers = newHashMap();
+
+ /**
+ * Writers that are currently in use
+ */
+ private final Set<SegmentBufferWriter> borrowed = newHashSet();
+
+ /**
+ * Retired writers that have not yet been flushed
+ */
+ private final Set<SegmentBufferWriter> disposed = newHashSet();
+
+ public GlobalSegmentBufferWriterPool(
+ @NotNull SegmentIdProvider idProvider,
+ @NotNull SegmentReader reader,
+ @NotNull String wid,
+ @NotNull Supplier<GCGeneration> gcGeneration) {
+ super(idProvider, reader, wid, gcGeneration);
+ }
+
+ @NotNull
+ @Override
+ public RecordId execute(@NotNull GCGeneration gcGeneration, @NotNull
WriteOperation writeOperation)
+ throws IOException {
+ SimpleImmutableEntry<?,?> key = new
SimpleImmutableEntry<>(currentThread(), gcGeneration);
+ SegmentBufferWriter writer = borrowWriter(key, gcGeneration);
+ try {
+ return writeOperation.execute(writer);
+ } finally {
+ returnWriter(key, writer);
+ }
+ }
+
+ @Override
+ public void flush(@NotNull SegmentStore store) throws IOException {
+ List<SegmentBufferWriter> toFlush = newArrayList();
+ List<SegmentBufferWriter> toReturn = newArrayList();
+
+ poolMonitor.enter();
+ try {
+ // Collect all writers that are not currently in use and clear
+ // the list so they won't get re-used anymore.
+ toFlush.addAll(writers.values());
+ writers.clear();
+
+ // Collect all borrowed writers, which we need to wait for.
+ // Clear the list so they will get disposed once returned.
+ toReturn.addAll(borrowed);
+ borrowed.clear();
+ } finally {
+ poolMonitor.leave();
+ }
+
+ // Wait for the return of the borrowed writers. This is the
+ // case once all of them appear in the disposed set.
+ if (safeEnterWhen(poolMonitor, allReturned(toReturn))) {
+ try {
+ // Collect all disposed writers and clear the list to mark
them
+ // as flushed.
+ toFlush.addAll(toReturn);
+ disposed.removeAll(toReturn);
+ } finally {
+ poolMonitor.leave();
+ }
+ }
+
+ // Call flush from outside the pool monitor to avoid potential
+ // deadlocks of that method calling SegmentStore.writeSegment
+ for (SegmentBufferWriter writer : toFlush) {
writer.flush(store);
}
- writers.clear();
- } finally {
- lock.writeLock().unlock();
+ }
+
+ /**
+ * Create a {@code Guard} that is satisfied if and only if {@link
#disposed}
+ * contains all items in {@code toReturn}
+ */
+ @NotNull
+ private Monitor.Guard allReturned(final List<SegmentBufferWriter>
toReturn) {
+ return new Monitor.Guard(poolMonitor) {
+
+ @Override
+ public boolean isSatisfied() {
+ return disposed.containsAll(toReturn);
+ }
+
+ };
+ }
+
+ /**
+ * Same as {@code monitor.enterWhen(guard)} but copes with that pesky
{@code
+ * InterruptedException} by catching it and setting this thread's
+ * interrupted flag.
+ */
+ private static boolean safeEnterWhen(Monitor monitor, Monitor.Guard
guard) {
+ try {
+ monitor.enterWhen(guard);
+ return true;
+ } catch (InterruptedException ignore) {
+ currentThread().interrupt();
+ return false;
+ }
+ }
+
+ /**
+ * Return a writer from the pool by its {@code key}. This method may
return
+ * a fresh writer at any time. Callers need to return a writer before
+ * borrowing it again. Failing to do so leads to undefined behaviour.
+ */
+ private SegmentBufferWriter borrowWriter(@NotNull Object key, @NotNull
GCGeneration gcGeneration) {
+ poolMonitor.enter();
+ try {
+ SegmentBufferWriter writer = writers.remove(key);
+ if (writer == null) {
+ writer = newWriter(gcGeneration);
+ }
+ borrowed.add(writer);
+ return writer;
+ } finally {
+ poolMonitor.leave();
+ }
+ }
+
+ /**
+ * Return a writer to the pool using the {@code key} that was used to
borrow
+ * it.
+ */
+ private void returnWriter(Object key, SegmentBufferWriter writer) {
+ poolMonitor.enter();
+ try {
+ if (borrowed.remove(writer)) {
+ checkState(writers.put(key, writer) == null);
+ } else {
+ // Defer flush this writer as it was borrowed while
flush() was called.
+ disposed.add(writer);
+ }
+ } finally {
+ poolMonitor.leave();
+ }
}
}
@NotNull
@Override
- public RecordId execute(@NotNull GCGeneration gcGeneration,
- @NotNull WriteOperation writeOperation)
- throws IOException {
- lock.readLock().lock();
- SegmentBufferWriter writer = getWriter(currentThread(), gcGeneration);
- try {
- return writeOperation.execute(writer);
- } finally {
- lock.readLock().unlock();
- }
+ public GCGeneration getGCGeneration() {
+ return gcGeneration.get();
}
- private SegmentBufferWriter getWriter(@NotNull Thread thread, @NotNull
GCGeneration gcGeneration) {
- SimpleImmutableEntry<?,?> key = new SimpleImmutableEntry<>(thread,
gcGeneration);
- SegmentBufferWriter writer = writers.get(key);
- if (writer == null) {
- writer = new SegmentBufferWriter(
- idProvider,
- reader,
- getWriterId(wid),
- gcGeneration
- );
- writers.put(key, writer);
- }
- return writer;
+ protected SegmentBufferWriter newWriter(@NotNull GCGeneration
gcGeneration) {
+ return new SegmentBufferWriter(idProvider, reader, getWriterId(),
gcGeneration);
}
- private String getWriterId(String wid) {
+ protected String getWriterId() {
if (++writerId > 9999) {
writerId = 0;
}
diff --git
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
index 099b3e53e0..d7c1a9f430 100644
---
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
+++
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/file/FileStore.java
@@ -39,13 +39,7 @@ import org.apache.jackrabbit.guava.common.base.Supplier;
import org.apache.jackrabbit.guava.common.io.Closer;
import
org.apache.jackrabbit.guava.common.util.concurrent.UncheckedExecutionException;
import org.apache.jackrabbit.oak.commons.Buffer;
-import org.apache.jackrabbit.oak.segment.RecordId;
-import org.apache.jackrabbit.oak.segment.Segment;
-import org.apache.jackrabbit.oak.segment.SegmentId;
-import org.apache.jackrabbit.oak.segment.SegmentNodeState;
-import org.apache.jackrabbit.oak.segment.SegmentNotFoundException;
-import org.apache.jackrabbit.oak.segment.SegmentNotFoundExceptionListener;
-import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.*;
import org.apache.jackrabbit.oak.segment.compaction.SegmentGCOptions;
import org.apache.jackrabbit.oak.segment.file.ShutDown.ShutDownCloser;
import org.apache.jackrabbit.oak.segment.file.cancel.Canceller;
@@ -145,7 +139,7 @@ public class FileStore extends AbstractFileStore {
this.segmentWriter = defaultSegmentWriterBuilder("sys")
.withGeneration(() -> getGcGeneration().nonGC())
- .withWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.GLOBAL)
.with(builder.getCacheManager()
.withAccessTracking("WRITE", statsProvider))
.build(this);
@@ -197,7 +191,7 @@ public class FileStore extends AbstractFileStore {
defaultSegmentWriterBuilder("c")
.with(builder.getCacheManager().withAccessTracking("COMPACT", statsProvider))
.withGeneration(generation)
- .withWriterPool()
+
.withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
.build(this)
);
diff --git
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/memory/MemoryStore.java
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/memory/MemoryStore.java
index 50b0ee99f7..916f4cab08 100644
---
a/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/memory/MemoryStore.java
+++
b/oak-segment-tar/src/main/java/org/apache/jackrabbit/oak/segment/memory/MemoryStore.java
@@ -27,17 +27,7 @@ import java.util.concurrent.ConcurrentMap;
import org.apache.jackrabbit.guava.common.collect.Maps;
import org.apache.jackrabbit.oak.commons.Buffer;
-import org.apache.jackrabbit.oak.segment.CachingSegmentReader;
-import org.apache.jackrabbit.oak.segment.Revisions;
-import org.apache.jackrabbit.oak.segment.Segment;
-import org.apache.jackrabbit.oak.segment.SegmentId;
-import org.apache.jackrabbit.oak.segment.SegmentIdFactory;
-import org.apache.jackrabbit.oak.segment.SegmentIdProvider;
-import org.apache.jackrabbit.oak.segment.SegmentNotFoundException;
-import org.apache.jackrabbit.oak.segment.SegmentReader;
-import org.apache.jackrabbit.oak.segment.SegmentStore;
-import org.apache.jackrabbit.oak.segment.SegmentTracker;
-import org.apache.jackrabbit.oak.segment.SegmentWriter;
+import org.apache.jackrabbit.oak.segment.*;
import org.apache.jackrabbit.oak.spi.blob.BlobStore;
import org.apache.jackrabbit.oak.stats.NoopStats;
import org.jetbrains.annotations.NotNull;
@@ -72,7 +62,9 @@ public class MemoryStore implements SegmentStore {
});
this.revisions = new MemoryStoreRevisions();
this.segmentReader = new CachingSegmentReader(this::getWriter, null,
16, 2, NoopStats.INSTANCE);
- this.segmentWriter =
defaultSegmentWriterBuilder("sys").withWriterPool().build(this);
+ this.segmentWriter = defaultSegmentWriterBuilder("sys")
+ .withWriterPool(SegmentBufferWriterPool.PoolType.GLOBAL)
+ .build(this);
revisions.bind(this);
segmentWriter.flush();
}
diff --git
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java
index ae6856a857..5ea1f24d79 100644
---
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java
+++
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/NodeRecordTest.java
@@ -105,7 +105,7 @@ public class NodeRecordTest {
SegmentWriter writer = defaultSegmentWriterBuilder("test")
.withGeneration(generation)
- .withWriterPool()
+ .withWriterPool(SegmentBufferWriterPool.PoolType.GLOBAL)
.with(nodesOnlyCache())
.build(store);
diff --git
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
index 8aea965f66..661a037876 100644
---
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
+++
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorExternalBlobTest.java
@@ -31,7 +31,7 @@ public class ParallelCompactorExternalBlobTest extends
AbstractCompactorExternal
protected ParallelCompactor createCompactor(@NotNull FileStore fileStore,
@NotNull GCGeneration generation) {
SegmentWriter writer = defaultSegmentWriterBuilder("c")
.withGeneration(generation)
- .withWriterPool()
+
.withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
.build(fileStore);
return new ParallelCompactor(
diff --git
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
index 596022bba3..627a3acf4a 100644
---
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
+++
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/ParallelCompactorTest.java
@@ -31,7 +31,7 @@ public class ParallelCompactorTest extends
AbstractCompactorTest {
protected ParallelCompactor createCompactor(@NotNull FileStore fileStore,
@NotNull GCGeneration generation) {
SegmentWriter writer = defaultSegmentWriterBuilder("c")
.withGeneration(generation)
- .withWriterPool()
+
.withWriterPool(SegmentBufferWriterPool.PoolType.THREAD_SPECIFIC)
.build(fileStore);
return new ParallelCompactor(
diff --git
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPoolTest.java
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPoolTest.java
index 6431ca7902..c7417e1469 100644
---
a/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPoolTest.java
+++
b/oak-segment-tar/src/test/java/org/apache/jackrabbit/oak/segment/SegmentBufferWriterPoolTest.java
@@ -29,7 +29,8 @@ import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertTrue;
import java.io.IOException;
-import java.util.concurrent.Callable;
+import java.util.Arrays;
+import java.util.List;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
@@ -40,10 +41,12 @@ import
org.apache.jackrabbit.oak.segment.WriteOperationHandler.WriteOperation;
import org.apache.jackrabbit.oak.segment.file.tar.GCGeneration;
import org.apache.jackrabbit.oak.segment.memory.MemoryStore;
import org.jetbrains.annotations.NotNull;
-import org.jetbrains.annotations.Nullable;
import org.junit.After;
import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.junit.runners.Parameterized;
+@RunWith(Parameterized.class)
public class SegmentBufferWriterPoolTest {
private final MemoryStore store = new MemoryStore();
@@ -51,17 +54,21 @@ public class SegmentBufferWriterPoolTest {
private GCGeneration gcGeneration = GCGeneration.NULL;
- private final SegmentBufferWriterPool pool = new SegmentBufferWriterPool(
- store.getSegmentIdProvider(),
- store.getReader(),
- "",
- () -> gcGeneration
- );
+ private final SegmentBufferWriterPool pool;
private final ExecutorService[] executors = new ExecutorService[] {
newSingleThreadExecutor(), newSingleThreadExecutor(),
newSingleThreadExecutor()};
- public SegmentBufferWriterPoolTest() throws IOException { }
+ @Parameterized.Parameters
+ public static List<SegmentBufferWriterPool.PoolType> poolTypes() {
+ return Arrays.asList(SegmentBufferWriterPool.PoolType.values());
+ }
+
+ public SegmentBufferWriterPoolTest(SegmentBufferWriterPool.PoolType
poolType) throws IOException {
+ pool = SegmentBufferWriterPool.factory(
+ store.getSegmentIdProvider(), store.getReader(), "", () ->
gcGeneration)
+ .newPool(poolType);
+ }
@After
public void tearDown() {
