WIP: add wrapper around AsyncProcessor to enable async processing engine only with async threads configured #306
Conversation
| private final String logPrefix; | ||
| private final Logger log; | ||
|
|
||
| private final String streamThreadName; | ||
| private final String asyncProcessorName; | ||
| private final TaskId taskId; | ||
|
|
||
| private final AsyncThreadPool threadPool; | ||
| private final SchedulingQueue<KIn> schedulingQueue; | ||
| private final FinalizingQueue finalizingQueue; |
There was a problem hiding this comment.
The nice thing about this change is that we only instantiate the AsyncProcessor after the processor is initialized, so we can actually make all these variables final. I was hoping to do this anyway
| @@ -69,12 +66,18 @@ | |||
| * -Coordinates the handoff of records between the StreamThread and AyncThreads | |||
| * -The starting and ending point of all async events -- see {@link AsyncEvent} | |||
| */ | |||
| public class AsyncProcessor<KIn, VIn, KOut, VOut> | |||
| implements Processor<KIn, VIn, KOut, VOut>, FixedKeyProcessor<KIn, VIn, VOut> { | |||
| public class AsyncProcessor<KIn, VIn, KOut, VOut> { | |||
There was a problem hiding this comment.
Since this no longer implements the Processor interfaces, I think I'll rename it to "AsyncProcessingEngine" or something to that effect (suggestions welcome). And then rename the new wrapper class that creates/delegates to this class to "AsyncProcessor". But I'll wait until the PR has been reviewed/approved to do the renamings since that might impact a bunch of code
| * 1. deduplicating and delegating code between the regular and fixed-key processor versions | ||
| * 2. disabling async processing altogether if the thread pool size is 0 | ||
| */ | ||
| public class MaybeAsyncProcessor<KIn, VIn, KOut, VOut> |
There was a problem hiding this comment.
not the final name for this -- plan to rename it to "AsyncProcessor" or similar. See this comment
| public AsyncProcessor<KIn, VIn, KIn, VOut> get() { | ||
| return createAsyncFixedKeyProcessor(userProcessorSupplier.get(), asyncStoreBuilders); | ||
| public MaybeAsyncProcessor<KIn, VIn, KIn, VOut> get() { | ||
| return MaybeAsyncProcessor.createFixedKeyProcessor(userProcessorSupplier.get(), asyncStoreBuilders); |
There was a problem hiding this comment.
the one thing I haven't implemented yet is that we initialize the store builders as AsyncStoreBuilders right from the start. I still need to push that back so that we don't end up wrapping the user's store suppleirs in the async stuff if that's not enabled
|
Shelving this for now since the approach becomes rather messy once we fix the AsyncStoreBuilder issue (essentially we'll need to provide wrapper classes for every single async store type, which are already wrappers for each of the actual StateStore types) Will probably pick this back up again at some point in the future, but in the meantime we advise users to simply branch their DSL operators based on the configuration in their own code. For example: |
Add a wrapper around the AsyncProcessor class and only instantiate it if the user has configured a nonzero thread pool size. This allows turning async on/off with just a config, without needing to add/remove the wrapper around the ProcessorSupplier as well.