Add rd_kafka_destroy_flags() #1932
Conversation
…lbacks If an rko was enqueued on a forwarded queue who's final destination queue was disabled (but not the initial one), the rko would be set up with the orignal queue's rko_serve callback which would then be triggered on the reply queue with the rko type changed to a REPLY with an error set, something these callbacks will not gracefully handle (and shouldn't need to). This fix makes sure the original dest queue rko_serve callback is only set on the rko if it is going to be successfully enqueued.
There are currently two flags: * RD_KAFKA_DESTROY_F_IMMEDIATE - allow an application to speed up termination at the risk of leaking memory and resources. This is a best-effort cleanup. * RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE - don't leave group, commit final offsets, trigger rebalance, from rd_kafka_destroy*() if consumer_close() was not explicitly called priorily.
| * @remark rd_kafka_consumer_close() will be called from this function | ||
| * if the instance type is RD_KAFKA_CONSUMER, a \c group.id was | ||
| * configured, and the rd_kafka_consumer_close() was not | ||
| * explicitly called by the application. This in turn may |
There was a problem hiding this comment.
more important to note is it may trigger broker requests (callbacks may be triggered regardless).
| * | ||
| */ | ||
| RD_EXPORT | ||
| void rd_kafka_destroy_flags (rd_kafka_t *rk, int flags); |
There was a problem hiding this comment.
i guess this name is ok. the Ex extension used by MS has the advantage that most people looking at it will know it's a new improved version of an older method by convention whereas the word flags is specific to this case. it also potentially reads as 'destroying the flags'. On the other hand, i'm not confused.
There was a problem hiding this comment.
Right, I thought about "destroying the flags", but the librdkafka naming scheme is "rd_kafka__destroy", e.g., rd_kafka_topic_destroy, so I think destroy_flags is okay.
|
lgtm with the disclaimer that the logic in this PR is difficult to review properly given limited familiarity with the code. |
From Dispose() you would call
rd_kafka_destroy_flags(RD_KAFKA_DESTROY_F_NO_CONSUMER_CLOSE).This should probably be changed in the Python destructor as well.