Make it possible to react on rebalance #27
Conversation
|
Thanks for this pull request. This is definitely a feature to have. Taking the time to review it. |
|
Looks good to me. I'm just wondering how to test it. It would also be good to have it working with Lwt. |
|
It will be more complex to have this working for Lwt which uses the former batch API. |
| } | ||
|
|
||
| CAMLprim value ocaml_kafka_async_new_consumer(value caml_rebalance_callback, value caml_consumer_options) { | ||
| CAMLparam2(caml_rebalance_callback, caml_consumer_options); | ||
| CAMLlocal1(result); |
There was a problem hiding this comment.
| CAMLlocal1(result); | |
| CAMLlocal2(caml_callback, result); |
| if (Is_block(caml_rebalance_callback)) { | ||
| opaque = ocaml_kafka_opaque_create(caml_rebalance_callback); |
There was a problem hiding this comment.
| if (Is_block(caml_rebalance_callback)) { | |
| opaque = ocaml_kafka_opaque_create(caml_rebalance_callback); | |
| if (Is_block(caml_rebalance_callback)) { | |
| caml_callback = Field(caml_rebalance_callback, 0); | |
| opaque = ocaml_kafka_opaque_create(caml_callback); |
| caml_cb = ((ocaml_kafka_opaque*)opaque)->caml_callback; | ||
|
|
||
| // do the default thing, but also call our OCaml code | ||
| switch (err) |
There was a problem hiding this comment.
this should probably handle cooperative assignors too, and call rd_kafka_incremental_assign (see https://docs.confluent.io/platform/current/clients/librdkafka/html/rdkafka_8h.html#a10db731dc1a295bd9884e4f8cb199311)
| caml_callback(caml_cb, Val_unit); | ||
| break; | ||
|
|
||
| default: |
There was a problem hiding this comment.
this is missing a call to rd_kafka_assign():
In this latter case (arbitrary error), the application must call rd_kafka_assign(rk, NULL) to synchronize state.
I am attempting to implement batching using the modern Kafka API and for this I need to do a few steps as described in this comment by the author of librdkafka. One of them is to be able to flush the messages that I have collected so far when a rebalance has happened. This adds (minimal) support to notify the OCaml code when this has happened.
Let me know what you think. In particular this does not (yet?) allow the OCaml code to influence the partitioning, it just does essentially what the documentation gives example of a callback.