alexeykudinkin commented on code in PR #7769:
URL: https://github.com/apache/hudi/pull/7769#discussion_r1091314159
##########
hudi-client/hudi-client-common/src/main/java/org/apache/hudi/util/ExecutorFactory.java:
##########
@@ -33,31 +33,47 @@
public class ExecutorFactory {
- public static <I, O, E> HoodieExecutor<E> create(HoodieWriteConfig
hoodieConfig,
+ public static <I, O, E> HoodieExecutor<E> create(HoodieWriteConfig config,
Iterator<I> inputItr,
HoodieConsumer<O, E>
consumer,
Function<I, O>
transformFunction) {
- return create(hoodieConfig, inputItr, consumer, transformFunction,
Functions.noop());
+ return create(config, inputItr, consumer, transformFunction,
Functions.noop());
}
- public static <I, O, E> HoodieExecutor<E> create(HoodieWriteConfig
hoodieConfig,
+ public static <I, O, E> HoodieExecutor<E> create(HoodieWriteConfig config,
Iterator<I> inputItr,
HoodieConsumer<O, E>
consumer,
Function<I, O>
transformFunction,
Runnable
preExecuteRunnable) {
- ExecutorType executorType = hoodieConfig.getExecutorType();
-
+ ExecutorType executorType = config.getExecutorType();
switch (executorType) {
case BOUNDED_IN_MEMORY:
- return new
BoundedInMemoryExecutor<>(hoodieConfig.getWriteBufferLimitBytes(), inputItr,
consumer,
+ return new
BoundedInMemoryExecutor<>(config.getWriteBufferLimitBytes(), inputItr, consumer,
transformFunction, preExecuteRunnable);
case DISRUPTOR:
- return new
DisruptorExecutor<>(hoodieConfig.getWriteExecutorDisruptorWriteBufferSize(),
inputItr, consumer,
- transformFunction,
hoodieConfig.getWriteExecutorDisruptorWaitStrategy(), preExecuteRunnable);
+ return new
DisruptorExecutor<>(config.getWriteExecutorDisruptorWriteBufferSize(),
inputItr, consumer,
+ transformFunction, config.getWriteExecutorDisruptorWaitStrategy(),
preExecuteRunnable);
case SIMPLE:
return new SimpleExecutor<>(inputItr, consumer, transformFunction);
default:
throw new HoodieException("Unsupported Executor Type " + executorType);
}
}
+
+ /**
+ * Checks whether configured {@link HoodieExecutor} buffer records (for ex,
by holding them
+ * in the queue)
+ */
+ public static boolean isBufferingRecords(HoodieWriteConfig config) {
Review Comment:
I actually realized that this is not possible unfortunately: we're copying
in transformers which we pass as args to ctor of the respective Executor,
therefore we can't just call a method on it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
