Hi everyone! Does anybody know how to interrupt librdkafka internal threads using a signal and let the application process some signal? I have daemon+worker application and kafka is used inside the worker. When daemon receives SIGTERM it sends one to all workers. The problem is when the application is consuming a topic, the signal handler inside worker is not called. However using strace I can see, that process actually receives SIGTERM and hangs inside system calls like this:
--- SIGTERM {si_signo=SIGTERM, si_code=SI_USER, si_pid=30489, si_uid=1000} ---
rt_sigreturn({mask=~[KILL TERM STOP RTMIN RT_1]}) = -1 EINTR (Interrupted system call)
futex(0x5602a96ca6bc, FUTEX_WAIT_BITSET_PRIVATE|FUTEX_CLOCK_REALTIME, 17, {1484151715, 231482000}, ffffffff) = -1 ETIMEDOUT (Connection timed out)
futex(0x5602a96ca690, FUTEX_WAKE_PRIVATE, 1) = 0
rt_sigprocmask(SIG_BLOCK, ~[RTMIN RT_1], ~[KILL TERM STOP RTMIN RT_1], 8) = 0
rt_sigprocmask(SIG_SETMASK, ~[KILL TERM STOP RTMIN RT_1], NULL, 8) = 0
rt_sigprocmask(SIG_BLOCK, ~[RTMIN RT_1], ~[KILL TERM STOP RTMIN RT_1], 8) = 0
rt_sigprocmask(SIG_SETMASK, ~[KILL TERM STOP RTMIN RT_1], NULL, 8) = 0Lines after rt_sigreturn are repeated until the worker is killed after timeout.
I'm using asynchronous signal processing from php 7.1 and when I replace kafka consuming with sleep function everything works well.
Environment: php 7.1, php-rdkafka 3.0.0, librdkafka 0.9.2-237-gd3dcc0 (d3dcc0198517160b9c8e374da2e963f563eb2c6f)
@edenhill Yes. client send login and passwd, i process this information and make unique token and after this i need to send token to this user. But if i send this data to "consumer group" all consumers in this group get this info. Maybe i can create personal "consumer group" for every user but i think 100 000+ groups is bad idea..
thanks @edenhill - so I'm not doing any ack at this stage - this is simply a "fire and forget" function - I just want to dump stuff into Kafka from PHP. I have altered the socket.blocking.max.ms down to 100 and still the same, plus we can't use the SIGIO stuff because it's php-fpm. I'll look at the queue.buffering.max.ms now
it's literally logging the request information into Kafka from the app, one big JSON string, nothing complicated
I have adjusted the queue.buffering.max.ms and it has made a big difference, thanks @edenhill ! - I'll monitor now and see how it goes, but really appreciate that - is there any way I can see if a message has failed to send - like the config values I'm changing wont alter the fact that it'll still send the message right? if that makes sense
if you use acks=0 (which you typically dont want to do, hello message loss) then the delivery report callback will be triggered as soon as the message has been sent to the broker.
could it be because I'm using the stubs? and maybe the stubs are outdated?
@edenhill can you spare time to look at a bug report that I'm about to file with librdkafka ? (seg fault) Happening with PHP 5.6 and PHP7
..................FFF........S.SS..[New Thread 0x7fffdfccf700 (LWP 20884)]
[Thread 0x7fffdfccf700 (LWP 20884) exited]
[New Thread 0x7fffdfccf700 (LWP 20886)]
[Thread 0x7fffdfccf700 (LWP 20886) exited]
.......[New Thread 0x7fffdfccf700 (LWP 20887)]
[Thread 0x7fffdfccf700 (LWP 20887) exited]
[New Thread 0x7fffdfccf700 (LWP 20888)]
[Thread 0x7fffdfccf700 (LWP 20888) exited]
...........
Program received signal SIGSEGV, Segmentation fault.
0x00007fffec678a3b in rd_kafka_new () from /lib64/librdkafka.so.1
(gdb) quit
Sure! Will do
So you've patched it. It may be this. edenhill/librdkafka#850 - I'm getting a
bt so we know what issue we ran into though. To close this off :)
then there'll be a new feature release in a month or so
I will test this branch? https://github.com/edenhill/librdkafka/tree/0.9.3.x
thanks for the backtrace, it didnt say much though but that's fine, you probably identified the issue anyway.
// Global config settings - see: https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md#global-configuration-properties
$conf = new ConsumerConf();
$conf->set("enable.auto.commit", "false");
$conf->set("socket.blocking.max.ms", 1);
$conf->set("fetch.wait.max.ms", 10);
$conf->set('metadata.broker.list', implode(',', $brokers));
$conf->set('group.id', $this->generateGroupId($topicName, $partition));
$conf->setErrorCb(function ($kafka, $err, $reason) {
printf("Kafka error: %s (reason: %s)\n", rd_kafka_err2str($err), $reason);
});
$topicConf = new TopicConf();
$topicConf->set('auto.offset.reset', 'smallest');
$conf->setDefaultTopicConf($topicConf);
$consumer = new HighLevelConsumer($conf);
$consumer->assign([
new TopicPartition($topicName, $partition, RD_KAFKA_OFFSET_STORED)
]);
but that's up to you and your use-case