Batch Receive

[oguzhantortop]

Can we receive messages in batches? Normally pulsar has the ability to receive batch messages in a given quantity in a given timeout. If a topic does not receive enough messages then it consumes messages in a given time. Does a reactive client of yours works in this fashion or do you need additional development for this feature?

[majusko]

Yes, that is how the Pulsar works, but the configuration is kept on default values. Would you like to have it exposed to override batching policy?

[majusko]

    public static final BatchReceivePolicy DEFAULT_POLICY = new BatchReceivePolicy(
            -1, 10 * 1024 * 1024, 100, TimeUnit.MILLISECONDS)

[oguzhantortop]

Not quite sure if we are talking about the same thing. I was actually trying to say that pulsar consumer has a method called consumer.batchReceive(). This is effective for scenarios like batch inserting to database. I looked at your code a little bit the one with PulsarConsumer annotation and as far as I see it doesn't use consumer.batchReceive() instead it uses messageListener. And in that case you might receive msgs in batch but you don't process them in batch... I hope I can express it...

[majusko]

I see. Thanks for explaining the DB use case. You are right, this is a good idea to implement in the library. Thank you 🙌

[oguzhantortop]

If I will have some time I would like to contribute. Though I am not familiar with Spring Reactive elements I might help for the non reactive part. A new annotation like @PulsarBatchConsumer might be good...

[majusko]

Yes, definitely should be a separate annotation. However, this would be quite a lot of work as you would need to have a different subscription mechanism since after the batchReceive timeout or msg limit is reached, the subscription needs to be renewed. Also, many tests would need to be configured there, so please go ahead only if you are confident about how to implement it. I may also help with the implementation. 🤞

[oguzhantortop]

What I am thinking is to create an infinite loop that calls the consumer batchReceive method. And this infinite loop should run inside a separate thread or maybe we might use another feature of the spring framework. I will start to write but I am not sure if this is the right approach. What do you think about it?

[oguzhantortop]

Instead creating a new annotation I used your existing annotation by adding a batch attribute. default value is false when you set it to true than I iniate a new thread with infinite loop. Of course there are tons of things to do. I just committed it for you to take a look at it.
https://github.com/oguzhantortop/pulsar-java-spring-boot-starter btw I switched to pulsar 2.9.2 because of my current environment...

[oguzhantortop]

Hi,
I have added documentation for annotation attributes and fixed the magic strings problems and some code smell. Could you please have a look?

Also, we need to set batch receive policies. Should we put those values from annotation attributes or should we use a properties file?

[majusko]

Thank you, @oguzhantortop Much better now.

Few points:

1. You are swallowing exceptions there. You forgot to send all exceptions to the sink so users can subscribe on all errors happening during the subscription:

sink.tryEmitNext(new FailedMessage(e, consumer, msg));

msgs != null // can be connected with previous if statement

3. Please collect completable futures as we do with private List<Consumer> consumers; just adding into a list + creating a getter inside a class so people can debug the thread's state if needed.

4. try to separate the whole batching part of the method to separate method because it's quite a robust method now
   from if (holder.getAnnotation().batch()) { until the end of the if

5. please configure your formatting on 4-spaces tabs - you have it at 8 - I'm not using Eclipse, but it should be pretty easy to configure for the project - In IntelliJ, it would be under the Code Style part in preferences. It also seems you are using tabs instead of whitespace. Also, make sure you don't change part of code that was developed by other programmers.

6. Add Policy - just use annotation fields - no need to put global configuration since there is always a default policy.

7. Add a few test cases

8. Update a README.md

Once all of these points are solved, please feel free to create PR - it's much easier to review that way, and I believe there might be only a few last cosmetics, and then this can be merged. Really, excellent work!

[oguzhantortop]

For the 1. item FailedMessage class is not suitable with Messages class. So I created a new class like below:
private final Sinks.Many<FailedMessages> batchSink = Sinks.many().multicast().onBackpressureBuffer(Queues.SMALL_BUFFER_SIZE, false);

I will do the final formatting on IntelliJ because my Eclipse shows 4 spaces tab already anyway I tried to do it manually as much as possible. I completed items other than 7. and 8. After completing 7 and 8 I will create PR. Till that time if something comes to your mind please let me know...

[majusko]

I think it is better to have one sink for errors, so it's easier to write loggers and people that rely on this already don't need to change anything. So please extend the class used in the sink with the new attribute batchMessages with a new getter and keep everything as it is so it's not a breaking change.

[majusko]

Sorry for the formatting - maybe it's just a GitHub UI - will check that in IDE once you raise a PR.

[oguzhantortop]

Added documentation to batch consumer configuration. Planning to complete unit tests this weekend.

[oguzhantortop]

Hi, I also finished with unit tests.

[majusko]

Perfect, one last thing - 2 tests are failing - the hardcoded consumers and producers calculator. Please increase the values. Also wrote a comment to the PR. After this I will merge it :)

[oguzhantortop]

Can you control again? :)

[majusko]

Done, merged! 💯 Congrats 🎉 ... I will do a release soon as some other minor changes should go out asap.

[riritack]

How to control the time interval of each batch in batch consume

[oguzhantortop]

you need to set timeoutMillis attribute of consumer annotation. But as far as I know it is available on version 1.2.0.

[riritack]

ok,thanks