Add batching processor #139

ryamagishi · 2021-12-09T02:11:12Z

Motivation

Task batching is a common-pattern that many Decaton users often implement by their own.
- i.e. Batching several tasks of type T to List and process them at once. e.g. when downstream-DB supports batching I/O (which often very efficient)
- Batch-flushing should be done in time-based and size-based.
So it's better to provide BatchingProcessor built-in to meet the common needs

Related Issue: #128

CLAassistant · 2021-12-09T02:11:16Z

All committers have signed the CLA.

ryamagishi · 2021-12-09T02:17:17Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+    @Override
+    public void process(ProcessingContext<T> context, T task) throws InterruptedException {
+        BatchingTask<T> newTask = new BatchingTask<>(context.deferCompletion(), context, task);
+        boolean isInitialTask = windowedTasks.isEmpty();


The motivation is to make a decision before adding a Task, and to scheduleFlush after adding a Task.

ryamagishi · 2021-12-09T02:18:57Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+
+    /**
+     * *MUST* call {@link BatchingTask#completion}'s {@link DeferredCompletion#complete()} or
+     * {@link BatchingTask#context}'s {@link ProcessingContext#retry()} method.


As a result of worrying about the retry process, I left it to the implementation side, but I think that it should be improved...

Since this is the main method of this class that we expect users to implement, the document should be bit more informative IMO.

At least it should cover topics like what to do after complete processing batch of tasks. (call .completion().complete() or retry() for each right?),

what happens if an error is thrown, which threads may possibly call this method etc.

ocadaruma

Thanks for submitting a patch!
Added few early feedbacks.

ocadaruma · 2021-12-09T09:38:02Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+            return;
+        }
+        synchronized (windowedTasks) {
+            processBatchingTasks(windowedTasks);


Seems like clearing passed batchingTasks is also implementer's responsibility?
I guess it's not good design, as it could be error-prone.

Let's pass copy of windowedTasks and clear it?

Thank you for the advice! I fixed it.
1a2e2c2

ocadaruma · 2021-12-09T09:39:36Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+    }
+
+    private void flush() {
+        if (!windowedTasks.isEmpty()) {


Hm, is that intentional? (not windowedTasks.isEmpty()?)

Sorry... I made a mistake.
4fdf913

ocadaruma · 2021-12-09T09:42:14Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+
+    // visible for testing
+    Runnable flushTask() {
+        return this::flush;


Seems flushTask() isn't called periodically?

I was thinking about the following, but it was certainly difficult to understand, so I fixed it!

When the first task comes in, scheduleFlush() is called.

flush() empties windowsTasks.

When the next task comes in after flush(), scheduleFlush() is called because windowTasks is empty.
2ef1bc0

ryamagishi · 2021-12-27T04:02:35Z

I'm sorry for the late response 🙇
I fixed the github commented part, added comments and tests for BatchingProcessor and changed PR status from draft.
I would be grateful if you could review this when you have time.

ocadaruma

Sorry for delaying the review!

Reviewed 1st round. PTAL comments.

ocadaruma · 2021-12-29T01:44:51Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+ * Batch-flushing should be done in time-based and size-based.
+ * @param <T> type of task to batch
+ */
+abstract public class BatchingProcessor<T> implements DecatonProcessor<T> {


[nits] should be ordered as public abstract

I fixed it. 4471e71

ocadaruma · 2021-12-29T01:45:30Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+     * @param capacity size limit for this processor. Every time tasks’size reaches capacity,
+     * tasks in past before reaching capacity are pushed to {@link BatchingTask#processBatchingTasks(List)}.
+     */
+    public BatchingProcessor(long lingerMillis, int capacity) {


As the general practice, the constructor of abstract class should be marked as protected

I fixed it. f6f97ba

ocadaruma · 2021-12-29T02:10:32Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+    @Accessors(fluent = true)
+    public static class BatchingTask<T> {
+        @Getter(AccessLevel.NONE)
+        public final Completion completion;


Let's make all fields as private and provide getter explicitly?

Also I think getter for completion is necessary to complete the task

I fixed it. c668bb3

ocadaruma · 2021-12-29T02:23:12Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+        BatchingTask<T> newTask = new BatchingTask<>(context.deferCompletion(), context, task);
+        windowedTasks.add(newTask);
+        if (windowedTasks.size() >= this.capacity) {
+            flush();


I think this should be executed in scheduler thread to avoid unnecessary lock contention between Decaton's processor thread and scheduler thread

Hm, then the problem is at the time flush() called, windowedTasks may contain more tasks than batch size, it's also not desirable.

Ok, then what do you think about following strategy?

Instead of using single final windowedTasks, have non-final List<BatchingTask> for "current" batch.

When "current" batch exceeds batch size, submit it to flusher and substitute current batch with new List.

Decaton will pause consumption if there's too many pending tasks, so even with this strategy, heap usage will not grow indefinitely.

I'm sorry, I probably don't understand what you're saying...
I tried to correct it with my own understanding, but please point out if it is wrong 🙇
cb78a8b

ocadaruma · 2022-01-05T08:44:23Z

processor/src/test/java/com/linecorp/decaton/processor/processors/BatchingProcessorTest.java

+
+        processor.process(context, task1);
+        processor.process(context, task2);
+        Thread.sleep(lingerMs * 2); // doubling just to make sure flush completed in background


Using Thread.sleep to wait something happen makes test result flaky.
Let's use CountDownLatch to make sure the condition happens. (Referring CompactionProcessorTest may be helpful)

Thank you for your reference.
I fixed it so that the test result is stable as much as possible.
357f2fb

ryamagishi · 2022-01-13T14:38:58Z

Thank you for your polite review!
I have corrected all the comments I received.
Please review again.

ocadaruma · 2022-01-14T04:50:15Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+     * *MUST* call {@link BatchingTask#completion}'s {@link DeferredCompletion#complete()} or
+     * {@link BatchingTask#context}'s {@link ProcessingContext#retry()} method.
+     */
+    abstract void processBatchingTasks(List<BatchingTask<T>> batchingTasks);


As the general practice, let's make this method protected

I fixed it. 5ae0959

ocadaruma · 2022-01-14T06:56:03Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+    public void process(ProcessingContext<T> context, T task) throws InterruptedException {
+        BatchingTask<T> newTask = new BatchingTask<>(context.deferCompletion(), context, task);
+        windowedTasks.add(newTask);
+        if (windowedTasks.size() >= this.capacity) {


In flush-thread centralization and batch.size-bound point of view, looks okay in current impl.

But I suggested in #139 (comment) is something like below:

public abstract class BatchingProcessor<T> implements DecatonProcessor<T> { private List<BatchingTask<T>> currentBatch = new ArrayList<>(); private void scheduleFlush() { executor.schedule(this::periodicallyFlushTask, lingerMillis, TimeUnit.MILLISECONDS); } private void periodicallyFlushTask() { final List<BatchingTask<T>> batch; synchronized (this) { if (!currentBatch.isEmpty()) { batch = currentBatch; currentBatch = new ArrayList<>(); } else { batch = null; } } if (batch != null) { processBatchingTasks(batch); } scheduleFlush(); } public void process(ProcessingContext<T> context, T task) throws InterruptedException { synchronized (this) { if (currentBatch.size() >= batchSize) { final List<BatchingTask<T>> batch = currentBatch; executor.submit(() -> processBatchingTasks(batch)); currentBatch = new ArrayList<>(); } currentBatch.add(task); } } }

I guess it's more straightforward than maintaining single windowedTasks instance with calling subList...?

I think this should be executed in scheduler thread to avoid unnecessary lock contention between Decaton's processor thread and scheduler thread

Thank you! I finally understood the meaning of the above sentence.
executor.submit(() -> processBatchingTasks(batch)); is important.
I fixed it. 9e35f1b

ryamagishi · 2022-01-20T21:06:43Z

#139 (comment)
Thanks for your test code advice.
While trying, I noticed that the abstract class mock wasn't working as expected.
So I fixed it to write test code simply without mocking. 87e11bd
ref: https://www.baeldung.com/junit-test-abstract-class

ryamagishi · 2022-02-22T00:03:02Z

@ocadaruma @kawamuray
Sorry for delaying the reply.
I've been trying to fix the integrationTest for a week or two, but I couldn't solve it...
I need to understand how ProcessorTestSuite works, but that's beyond my understanding😭
I'll continue to make efforts after this comment, but I'd appreciate it if you could tell me why the testBatchingProcessor_capacity fails. (Even if I extend timeout, it will time out.)

I have corrected other comments I received.
As soon as this PR is merged, I will add a documentation like task-compaction.adoc in another PR.

ocadaruma · 2022-03-02T00:52:57Z

Thanks for the fix.

why the testBatchingProcessor_capacity fails

I guess it's simply because linger-based flush is disabled on that test?
So, unless each processor instance receives exactly multiple of 100 (batch size) of tasks, few remaining tasks will be never get flushed.

ocadaruma · 2022-03-02T00:59:38Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+public abstract class BatchingProcessor<T> implements DecatonProcessor<T> {
+
+    private final ScheduledExecutorService executor;
+    private List<BatchingTask<T>> currentBatch = Collections.synchronizedList(new ArrayList<>());


Oh what I meant was for BatchingProcessorTest. (#139 (comment))
For this, every modification is done in synchronized block so not necessary to be wrapped I think.

I'm sorry, I misunderstood.
I fixed it.
08ed867, 242898b

ocadaruma · 2022-03-02T01:00:02Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+
+    @Value
+    @Accessors(fluent = true)
+    @RequiredArgsConstructor


@RequiredArgsConstructor is redundant as there's @Value

I fixed it.
aa51f27

This reverts commit a882452.

ryamagishi · 2022-03-20T08:00:24Z

I guess it's simply because linger-based flush is disabled on that test?
So, unless each processor instance receives exactly multiple of 100 (batch size) of tasks, few remaining tasks will be never get flushed.

Thank you for the advice.
I tried setting NUM_KEYS, NUM_SUBSCRIPTION_INSTANCES, NUM_PARTITIONS in ProcessorTestSuite to 1, but testBatchingProcessor_capacity() still failed...
I removed it for now.
ref: https://github.com/line/decaton/blob/master/testing/src/main/java/com/linecorp/decaton/testing/processor/ProcessorTestSuite.java#L102-L104
7821641

ryamagishi · 2022-03-20T08:01:45Z

Sorry for the late reply every time.
I will try to improve 🙇

ocadaruma · 2022-03-20T23:45:42Z

Thanks for the update!

but testBatchingProcessor_capacity() still failed

Ah maybe what I meant was more simple.

In testBatchingProcessor_capacity test, lingerMillis was set to Integer.MAX.
It means that batch-flushing is done by only size-based.
Let's see this by example. Let's say we set capacity to 2 and lingerMillis to Int.MAX. Then what happens we feed 3 tasks?
- 2 tasks will be processed because it will reach the capacity
- However, remaining 1 task will be never get flushed, so it causes consumption stuck.

I removed it for now.

However, I agree with removing it. Because what we want to check in integration test is BatchingProcessor's behavior in overall (as commented in #139 (comment)), so we don't have to write test scenario for each of linger=MAX, capacity=MAX I think.

ocadaruma · 2022-03-21T00:04:43Z

processor/src/main/java/com/linecorp/decaton/processor/processors/BatchingProcessor.java

+
+    /**
+     * Instantiate {@link BatchingProcessor}.
+     * If you only need one limit, please set large enough value to another.


As described in #139 (comment), we found that consumption will get stuck if we specify lingerMillis to Int.MAX.

Hm, so I think realistically we always specify non-Int.MAX value for both parameters.
Then let's just remove this line? To avoid mis-use of these parameters.

Exactly, I fixed it.
0cb57f7

ocadaruma · 2022-03-21T00:07:20Z

processor/src/it/java/com/linecorp/decaton/processor/BatchingProcessorTest.java

+        ProcessorTestSuite
+            .builder(rule)
+            .configureProcessorsBuilder(builder -> builder.thenProcess(
+                new BatchingProcessor<TestTask>(1000, Integer.MAX_VALUE) {


I suppose the purpose of this integration test is to verify the behavior of BatchingProcessor in overall (rather than checking if BatchingProcessor is implemented correctly for each individual cases like linger=Max, capacity=Max etc), so let's also specify non-Int.MAX value for capacity here?

I fixed it. (I renamed the test name too.)
0317258

ryamagishi · 2022-03-22T05:49:12Z

Thank you for your polite explanation every time!
If ProcessorTestSuite#numTasks = 10000 andBatchingProcessor#capacity = 100(and ProcessorTestSuite#NUM_SUBSCRIPTION_INSTANCES = 1), I thought it would succeed because the number is divisible.
However, I noticed that the number of tasks processed is actually not 10000. As you said, this doesn't work well.
I'm sorry that the pasted image is difficult to understand, my question has been solved.

ocadaruma · 2022-03-25T12:25:11Z

What I immediately thought is because of multiple partitions.
Since the partition count is set to 8 in ProcessorTestSuite, if we instantiate only 1 subscription instance, it will create 8 partition-processor. With partition.concurrency = 1 and ProcessorScope = THREAD, it will create 8 batching processor instances, and clearly the received tasks count will not be multiple of 100.

ocadaruma

LGTM.

Thanks for your great work!

@kawamuray Please check the change for your comment.

kawamuray

LGTM, thanks 👍

ryamagishi · 2022-03-29T02:18:53Z

~~May I merge it?~~
I don't have merge permission in the first place...

ryamagishi · 2022-03-29T02:25:34Z

As soon as this PR is merged, I will add a documentation like task-compaction.adoc in another PR!

ryamagishi added 3 commits December 6, 2021 15:09

Add BatchingProcessor

c5d715e

Remove unnecessary options

7fbfc78

Fix BatchingProcessor

05409e7

ryamagishi marked this pull request as draft December 9, 2021 02:12

ryamagishi commented Dec 9, 2021

View reviewed changes

ocadaruma reviewed Dec 9, 2021

View reviewed changes

kawamuray added the new feature Add a new feature label Dec 13, 2021

ryamagishi added 4 commits December 26, 2021 17:08

Fix wrong conditional branch

4fdf913

Fix to pass copy of windowedTasks and clear it

1a2e2c2

Fix to periodically call flush

2ef1bc0

Add test and comment to BatchingProcessor

1d78a1d

ryamagishi marked this pull request as ready for review December 27, 2021 03:53

ryamagishi requested a review from ocadaruma December 27, 2021 04:02

ocadaruma reviewed Jan 5, 2022

View reviewed changes

ryamagishi added 7 commits January 13, 2022 20:35

Fix order to public abstract

4471e71

Fix BatchingProcessor to protected

f6f97ba

Fix javadoc comment

85be0c3

Fix how to access BatchingTask's fileds

c668bb3

Fix BatchingProcessorTest

357f2fb

Fix indent

752509d

Fix BatchingProcessor's size management

cb78a8b

ocadaruma reviewed Jan 14, 2022

View reviewed changes

ryamagishi added 3 commits January 18, 2022 23:23

Fix testLingerLimit

87e11bd

Fix processBatchingTasks to protected

5ae0959

Fix BatchingProcessor

9e35f1b

ryamagishi added 7 commits February 22, 2022 06:08

Refactor BatchingProcessor

3475850

Wrap currentBatch with Collections.synchronizedList

a882452

Fix BatchingProcessor's doc

43f96d8

Add BatchingProcessor integrationTest

f3d7835

Fix javadoc exception

e1aee85

Fix BatchingProcessorTest for compileItJava

c6f20e1

Refactor BatchingProcessor to use namedThreadFactory

f6da8c6

ocadaruma reviewed Mar 2, 2022

View reviewed changes

ryamagishi added 4 commits March 20, 2022 15:10

Revert "Wrap currentBatch with Collections.synchronizedList"

08ed867

This reverts commit a882452.

Fix BatchingProcessorTest

242898b

Remove redundant @requiredargsconstructor

aa51f27

Remove testBatchingProcessor_capacity

7821641

ocadaruma reviewed Mar 21, 2022

View reviewed changes

ryamagishi added 2 commits March 22, 2022 13:43

Remove a confusing comment line.

0cb57f7

Fix integration BatchingProcessorTest

0317258

ryamagishi force-pushed the add-batching-processor branch from 40d355b to 0317258 Compare March 22, 2022 05:19

ocadaruma approved these changes Mar 25, 2022

View reviewed changes

kawamuray approved these changes Mar 27, 2022

View reviewed changes

ocadaruma merged commit 596ea76 into line:master Mar 29, 2022

ryamagishi mentioned this pull request Apr 10, 2022

Add task batching doc #150

Merged

ocadaruma mentioned this pull request Aug 20, 2024

Provide BatchingProcessor as a built-in processor #128

Closed

Add batching processor #139

Add batching processor #139

Conversation

ryamagishi commented Dec 9, 2021 • edited Loading

Motivation

CLAassistant commented Dec 9, 2021 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ocadaruma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryamagishi commented Dec 27, 2021

ocadaruma left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryamagishi commented Jan 13, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryamagishi commented Jan 20, 2022 • edited Loading

ryamagishi commented Feb 22, 2022

ocadaruma commented Mar 2, 2022

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryamagishi commented Mar 20, 2022

ryamagishi commented Mar 20, 2022

ocadaruma commented Mar 20, 2022 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

ryamagishi commented Mar 22, 2022

ocadaruma commented Mar 25, 2022

ocadaruma left a comment

Choose a reason for hiding this comment

kawamuray left a comment

Choose a reason for hiding this comment

ryamagishi commented Mar 29, 2022 • edited Loading

ryamagishi commented Mar 29, 2022

ryamagishi commented Dec 9, 2021 •

edited

Loading

CLAassistant commented Dec 9, 2021 •

edited

Loading

ryamagishi commented Jan 20, 2022 •

edited

Loading

ocadaruma commented Mar 20, 2022 •

edited

Loading

ryamagishi commented Mar 29, 2022 •

edited

Loading