all My Project use reactor kafka version 1.1.0 RELEASE and reactor-core 3.2.1. I want to consume slowly so I use delayElement, like this.
ReceiverOptions<String, String> options = receiverOptions.subscription(Collections.singleton(topic)) // kafka.consumer.upsert.topic
.addAssignListener(partitions -> log.debug("onPartitionsAssigned {}", partitions))
.addRevokeListener(partitions -> log.debug("onPartitionsRevoked {}", partitions));
KafkaReceiver.create(options).receive()
.delayElements(Duration.ofNanos(800000L))
.doOnError(this::onError)
.subscribe(this::onMessage);
I intend to consume about 1200 data per second. However, it takes a very short time to consume data. Maybe micro sec? so I change Duration to 1 millisec. so It was worked consume about 1000 data per second.
I add My test code here.
ArrayList<Integer> data = new ArrayList<>();
for (int i = 0; i <= 5000; i++) {
data.add(i);
}
log.info("start");
Flux.fromIterable(data)
//.doOnNext(t -> log.info("doOnNext : {}", t.toString()))
.delayElements(Duration.ofNanos(99999))
.subscribe(n -> {
//log.info("data : {}", n);
if (n == 5000) {
log.info("finish");
}
});
//duration is 0.99ms It take very short time.
2022-09-21T12:50:04.961+09:00 INFO --- [ main] c.g : start
2022-09-21T12:50:05.104+09:00 INFO --- [ parallel-9] c.g : finish
//duration is 1ms It take very long time.
2022-09-21T12:52:52.048+09:00 INFO --- [ main] c.g : start
2022-09-21T12:54:08.264+09:00 INFO --- [ parallel-9] c.g : finish
I searched for the above phenomenon, but I couldn't find it. Has anyone experienced the same phenomenon as me?
plz help me. Thank you.