Add files batching capabilities to the Policies #59

Symbianx · 2020-05-20T12:10:25Z

We use the SimplePolicy to restore events from S3 (1 event per file) and noticed messages took a lot of time before they were pushed into the topics.

This PR adds a new configuration option policy.batch_size that allows the connector to handle a maximum number of files at a time as configured.

Fixes #58.

coveralls · 2020-05-20T12:20:01Z

Coverage increased (+0.1%) to 95.697% when pulling 04be206 on Symbianx:add-simple-batch-policy into 588d310 on mmolimar:master.

mmolimar · 2020-05-25T00:16:04Z

Many thanks!
I think we could implement this at policy level to take advantage of this functionality to all policies (actually it's a feature of the next minor version).
Also, there is a related PR #60 about this, is that ok?

Symbianx · 2020-05-25T09:39:35Z

I went through the #60 and from my understanding it doesn't solve the same challenge as this one.
It does had chunking but still reads all files/messages before returning them for publishing and commiting.

I'm gonna try to refactor this to work at the AbstractPolicy level.

mmolimar · 2020-05-25T13:31:22Z

Great @Symbianx!
By the way, when you update the PR, could do the PR against develop` branch? I'd like to include all new features there and then I'll update the master branch with the new version released.

Symbianx · 2020-05-25T16:16:26Z

@mmolimar Made the change to setup the batching in the AbstractPolicy like you requested.

Symbianx · 2020-05-26T13:31:55Z

@mmolimar been trying to fix the failing test but with no luck. Can't even reproduce on my machine, you have any idea of what could be wrong?

mmolimar

Thanks @Symbianx
I added some comments.

src/main/java/com/github/mmolimar/kafka/connect/fs/policy/AbstractPolicy.java

mmolimar · 2020-05-27T00:48:41Z

src/main/java/com/github/mmolimar/kafka/connect/fs/policy/AbstractPolicy.java

@@ -105,16 +109,25 @@ private String convert(String uri) {
        if (hasEnded()) {
            throw new IllegalWorkerStateException("Policy has ended. Cannot be retried.");
        }
+
+        if (batchSize > 0 && currentIterator != null && currentIterator.hasNext()) {


I think it's better to avoid this check and set the files iterator with the previous value

If we skip this check, the preCheck and postCheck methods will be executed which will cause the CronPolicy, SleepyPolicy and HdfsFileWatcherPolicy to sleep between batches which breaks the current behaviour of sleeping only after handling all the files. Is this what you want?

Makes totally sense but just returning previous (in case it has elements and ignoring the batch size), ok?

That sounds ok.
Since previous is the original iterator (not BatchIterator) I will need to create a custom class extending Iterator with a method to reset the counter to 0. Then we can store the BatchIterator in the previous variable instead.

I will change it tomorrow.

Changed it to use com.google.common.collect.Iterators.partition instead which simplified the implementation and becomes more standard.

mmolimar · 2020-05-27T00:50:22Z

src/main/java/com/github/mmolimar/kafka/connect/fs/util/BatchIterator.java

+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class BatchIterator {


We could add this functionality inside the listFiles method in the AbstractPolicy class

If we put it in the listFiles method we will be batching each file system independently. Feels to me that is not the intended behaviour.

Ok. Maybe renaming this class to Iterators with a partition method with the batch size and a "duplicates" flag to allow removing duplicates in the same partition.

Sorry, I didn't understand what you mean by "allow removing duplicates in the same partition".

For example: in the first execution of the policy you get an iterator with 5 records (with a batch size of 3) so 2 remaining. In the next iteration (let's say we're using the Cron policy) we concat the previous 2 with other 5 records. We re-split that iterator and "maybe" in this batch could be an item duplicated, so we should ignore this record and get another one.

I don't think that will not be inline with the present implementation.
An execution is only finished when the iterator is fully traversed, only after that we will sleep. If we mix files from the current iterator with the next one we will be creating a weird behaviour in the policies.

mmolimar · 2020-06-11T16:09:42Z

Thanks!

Symbianx mentioned this pull request May 20, 2020

Batching Support #58

Closed

Symbianx force-pushed the add-simple-batch-policy branch from abcb768 to 04be206 Compare May 20, 2020 14:36

Symbianx added 2 commits May 25, 2020 17:56

Add a simple batch policy for file systems with a lot of files

5b4147e

Improve code test coverage

f1620ce

Symbianx force-pushed the add-simple-batch-policy branch from dec8c77 to 45c0709 Compare May 25, 2020 15:56

Symbianx changed the base branch from master to develop May 25, 2020 15:57

Symbianx force-pushed the add-simple-batch-policy branch 2 times, most recently from 81f6e1e to 21175ba Compare May 25, 2020 16:03

Setup batching in the AbstractPolicy instead of a seperate Policy.

7d1e4e9

Symbianx force-pushed the add-simple-batch-policy branch from 21175ba to 7d1e4e9 Compare May 25, 2020 16:04

Symbianx changed the title ~~Add a simple batch policy for file systems with a lot of files~~ Add files batching capabilities to the Policies May 25, 2020

Symbianx added 2 commits May 25, 2020 18:33

Fix unit tests

fd4e5dd

Fix unit tests by isolating newly created policies

e40faad

mmolimar reviewed May 27, 2020

View reviewed changes

Symbianx added 2 commits May 27, 2020 15:06

Rename iterator

4252273

Use com.google.common.collect.Iterators.partition method instead

02dd5a4

Symbianx force-pushed the add-simple-batch-policy branch from 4327185 to 02dd5a4 Compare June 2, 2020 17:37

Minor changes

4451c04

mmolimar merged commit 7a35c9b into mmolimar:develop Jun 11, 2020

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add files batching capabilities to the Policies #59

Add files batching capabilities to the Policies #59

Symbianx commented May 20, 2020 •

edited

Loading

coveralls commented May 20, 2020 •

edited

Loading

mmolimar commented May 25, 2020

Symbianx commented May 25, 2020 •

edited

Loading

mmolimar commented May 25, 2020

Symbianx commented May 25, 2020

Symbianx commented May 26, 2020

mmolimar left a comment

mmolimar May 27, 2020

Symbianx May 27, 2020

mmolimar May 27, 2020

Symbianx May 27, 2020

Symbianx Jun 2, 2020

mmolimar May 27, 2020

Symbianx May 27, 2020

mmolimar May 27, 2020

Symbianx May 27, 2020

mmolimar May 29, 2020

Symbianx Jun 2, 2020

mmolimar commented Jun 11, 2020

Add files batching capabilities to the Policies #59

Add files batching capabilities to the Policies #59

Conversation

Symbianx commented May 20, 2020 • edited Loading

coveralls commented May 20, 2020 • edited Loading

mmolimar commented May 25, 2020

Symbianx commented May 25, 2020 • edited Loading

mmolimar commented May 25, 2020

Symbianx commented May 25, 2020

Symbianx commented May 26, 2020

mmolimar 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

mmolimar commented Jun 11, 2020

Symbianx commented May 20, 2020 •

edited

Loading

coveralls commented May 20, 2020 •

edited

Loading

Symbianx commented May 25, 2020 •

edited

Loading