nginx worker processes exit with signal 11 when publishing messages in nginx #56

xianping · 2014-01-09T09:28:25Z

Hi edenhill,
I write c code which called by lua to produce messages to kafka server . The lua code is called in nginx server.
I create one kafka handle and add a list of brokers once in nginx initialisation. Then when receiving requests to nginx server, call the lua code to publish messages.
But I got the nginx worker processes exit with signal 11 once requests sent out.
I also wonder if it's right to hold one kafka handle for dealing with all topic write effectively.

As I don't know how to attach the code，i past the major code snippt below:
//the code try to create one kafka handle and add brokers
int mercury_add_kafkabroker(mercury_t *me, const char *kafka_brokers)
{

    if (!me || !kafka_brokers) return -1;

    init_error();

    char errstr[100];
    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set_error_cb(conf, err_cb);
    rd_kafka_conf_set_dr_cb(conf, msg_delivered);

    char err[100];
    rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    me->kafka_handler = rk; 

   if (rk == NULL) {
            set_error("rd_kafka_new fail");
            return -1;
    }

    /* Add brokers */
    if (rd_kafka_brokers_add(me->kafka_handler, kafka_brokers) == 0) {
            fprintf(stderr, "%% No valid brokers specified\n");
            return -1;
    }
    return 0;

}

//the code is called looply by receiving request to nginx server
int mercury_sendmsg_kafka(mercury_t *me, const char * topic, const char *msg)
{
if (!me || !msg) return -1;

    init_error();

    size_t msg_len = strlen(msg);

    char *opbuf = malloc(msg_len + 1);
    strcpy(opbuf, msg);

    if (NULL == me->kafka_topic_handler)  {
    rd_kafka_topic_conf_t * topic_conf = rd_kafka_topic_conf_new();
    rd_kafka_topic_t * rkt = rd_kafka_topic_new(me->kafka_handler, topic, topic_conf);
    me->kafka_topic_handler = rkt;
    }
    while(rd_kafka_produce(me->kafka_topic_handler, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_FREE, opbuf, msg_len, NULL, 0, NULL) == -1) {
        fprintf(stderr, "produce error; resend!");
        /* Poll to handle delivery reports */
        rd_kafka_poll(me->kafka_handler, 10); 
    }
    rd_kafka_poll(me->kafka_handler, 0); 

    if (NULL != me->kafka_handler)  { 
    rd_kafka_destroy(me->kafka_handler);
    }


    return 0;

}

Thanks
Aaron

The text was updated successfully, but these errors were encountered:

edenhill · 2014-01-09T09:37:54Z

Can you provide a backtrace of the crash by running it in gdb?

Are you really supposed to destroy me->kafka_handler in sendmsg_kafka()? I.e., is sendmsg only called once?
If the function is called again it will use the destroyed kafka_handler pointer, which will cause a crash.
Also, before destroying your kafka handle you will need to make sure that all messages have been sent by checking
rd_kafka_outq_len(rk) == 0), something like:

while (rd_kafka_outq_len(rk) > 0)
    rd_kafka_poll(rk, 10);

xianping · 2014-01-09T10:49:04Z

Based on your suggestion, I modify the sendmsg code below( by using gdb find a stupid mistake which make process exit, I have solved it). The new issue is nothing sent out with below code. Could you give more advise?

int mercury_sendmsg_kafka(mercury_t *me, const char * topic, const char *msg)
{
if (!me || !msg) return -1;

init_error();

size_t msg_len = strlen(msg);

char *opbuf = malloc(msg_len + 1);
strcpy(opbuf, msg);

if (NULL == me->kafka_topic_handler)  {
rd_kafka_topic_conf_t * topic_conf = rd_kafka_topic_conf_new();
rd_kafka_topic_t * rkt = rd_kafka_topic_new(me->kafka_handler, topic, topic_conf);
me->kafka_topic_handler = rkt;
}
while(rd_kafka_produce(me->kafka_topic_handler, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_FREE, opbuf, msg_len, NULL, 0, NULL) == -1) {
    fprintf(stderr, "produce error; resend!");

    }
     /* Poll to handle delivery reports */
  while (rd_kafka_outq_len(me->kafka_handler) > 0){ //add here...

   //program can hit here repeatly.......
    rd_kafka_poll(me->kafka_handler, 10); 
    }


return 0;

}

edenhill · 2014-01-09T10:56:56Z

Could you print errno if rd_kafka_produce() fails?

fprintf(stderr, "produce error: %s: resend\n", strerror(errno));

also, try enabling rdkafka debugging by setting conf property "debug" to "all":

rd_kafka_conf_set(conf, "debug", "all", errstr, sizeof(errstr));

xianping · 2014-01-10T02:18:55Z

xianping · 2014-01-10T09:56:31Z

edenhill,
I wonder if my usage in nginx is right. Or if you have examples. I tested by using one kafka server and fixed partition with 0, still encounter 'no partition' error.
thanks for help

edenhill · 2014-01-10T10:14:16Z

Can you check with the bin/kafka-topic.sh tool (official apache kafka package) that topic item.010001 exists?

Can you provide more/all debugging output?

xianping · 2014-01-10T13:36:12Z

As its weekend I will retest next Monday. Report here later

xianping · 2014-01-13T07:54:07Z

It seems that client can't get the leader partition.

Alternative experiment:
For each request in nginx, we create a new kafka handle, then add brokers and send message if a new topic arrives, we can produce message to the topic, but memory usage keeps hiking up, no release. So I wonder if this is right way to initialise handle and add brokers once in nignx. Could you give more guide?

edenhill · 2014-01-13T11:50:00Z

Do you have auto.create.topics.enable=true configured on your broker(s)?

Can you provide some more debug log output?

edenhill · 2014-01-13T11:58:13Z

Creating a new kafka handle for each new connection in nginx seems kind of expensive since it will:

connect to each of the provided/configured bootstrap brokers
request metadata from each bootstrap broker
connect once to each broker reported in the metadata info.

So there is at least B+N (B=bootstrap brokers, N=brokers) connections made and similar number of metadata requests before it will be able to start producing messages.

After that your message will be sent to one of the brokers and it will wait for a reply (if request.required.acks is not 0).

I dont know the internals of nginx, but would it be possible to share the kafka and topic handles between requests?
The rdkafka API is thread safe so you can easily produce from different threads.

xianping · 2014-01-14T09:20:37Z

Only the first record can be written to topic.
In kafka server.log, the bufferunderflowexception keep writing, see below:

[2014-01-14 17:06:54,232] INFO New leader is 2 (kafka.server.ZookeeperLeaderElector$LeaderChangeListener)
[2014-01-14 17:06:54,235] INFO [ReplicaFetcherManager on broker 2] Removing fetcher for partition test,0
[2014-01-14 17:06:54,283] INFO [Kafka Server 2], Started (kafka.server.KafkaServer)
[2014-01-14 17:06:54,297] INFO [ReplicaFetcherManager on broker 2] Removing fetcher for partition m.100000,0
[2014-01-14 17:06:54,299] INFO [Replica Manager on Broker 2]: Handled leader and isr request Name:LeaderAndIsrRequest;Version:0;Controller:2;ControllerEpoch:3;Correlati
onId:0;ClientId:id_2-host_127.0.0.1-port_9092;PartitionState:(test,0) -> (LeaderAndIsrInfo:(Leader:2,ISR:2,LeaderEpoch:0,ControllerEpoch:1),ReplicationFactor:1),AllRepl
icas:2),(m.100000,0) -> (LeaderAndIsrInfo:(Leader:2,ISR:2,LeaderEpoch:0,ControllerEpoch:1),ReplicationFactor:1),AllReplicas:2);Leaders:id:2,host:127.0.0.1,port:9092 (ka
fka.server.ReplicaManager)
[2014-01-14 17:09:27,015] ERROR Closing socket for /127.0.0.1 because of error (kafka.network.Processor)
java.nio.BufferUnderflowException
at java.nio.Buffer.nextGetIndex(Buffer.java:480)
at java.nio.HeapByteBuffer.getShort(HeapByteBuffer.java:285)
at kafka.api.ApiUtils$.readShortString(ApiUtils.scala:34)
at kafka.api.TopicMetadataRequest$$anonfun$readFrom$1.apply(TopicMetadataRequest.scala:44)
at kafka.api.TopicMetadataRequest$$anonfun$readFrom$1.apply(TopicMetadataRequest.scala:43)
at scala.collection.immutable.Range$ByOne$class.foreach(Range.scala:285)
at scala.collection.immutable.Range$$anon$2.foreach(Range.scala:265)
at kafka.api.TopicMetadataRequest$.readFrom(TopicMetadataRequest.scala:43)
at kafka.api.RequestKeys$$anonfun$4.apply(RequestKeys.scala:37)
at kafka.api.RequestKeys$$anonfun$4.apply(RequestKeys.scala:37)
at kafka.network.RequestChannel$Request.(RequestChannel.scala:49)
at kafka.network.Processor.read(SocketServer.scala:353)
at kafka.network.Processor.run(SocketServer.scala:245)
at java.lang.Thread.run(Thread.java:662)

BTW, I use one nginx process to test for simplicity. Why the resting records can't get topic metadata?

xianping · 2014-01-14T09:40:22Z

I modify my code for above test as follows:
int mercury_sendmsg_kafka(mercury_t *me, const char * topic, const char *msg)
{
if (!me || !msg) return -1;
mercury_add_kafkabroker(me, "127.0.0.1:9092");
fprintf(stderr,"mercury_sendmsg_kafka:%s\n ",topic);

    init_error();

    size_t msg_len = strlen(msg);

    char *opbuf = malloc(msg_len + 1);
    strcpy(opbuf, msg);

    rd_kafka_topic_conf_t * topic_conf = rd_kafka_topic_conf_new();
    rd_kafka_topic_t * rkt = rd_kafka_topic_new(me->kafka_handler, topic, topic_conf);
    me->kafka_topic_handler = rkt;

    while(rd_kafka_produce(me->kafka_topic_handler, RD_KAFKA_PARTITION_UA, RD_KAFKA_MSG_F_FREE, opbuf, msg_len, NULL, 0, NULL) == -1) {
        //fprintf(stderr, "produce error; resend!");
        //test xp...
        fprintf(stderr, "produce error: %s: resend\n", strerror(errno));
        /* Poll to handle delivery reports */

    }
    //rd_kafka_poll(me->kafka_handler, 0);
    while (rd_kafka_outq_len(me->kafka_handler) > 0){
          rd_kafka_poll(me->kafka_handler, 10);
    }
    if (NULL != me->kafka_topic_handler)  {
            rd_kafka_topic_destroy(me->kafka_topic_handler);
        //rd_kafka_destroy(me->kafka_handler);
    }


    return 0;

}

int mercury_add_kafkabroker(mercury_t *me, const char *kafka_brokers)
{
if(NULL != me->kafka_handler){
return 0;
}

    if (!me || !kafka_brokers) return -1;

    init_error();

    char errstr[100];
    char test_errstr[200];
    rd_kafka_conf_t *conf = rd_kafka_conf_new();
    rd_kafka_conf_set_error_cb(conf, err_cb);
    rd_kafka_conf_set_dr_cb(conf, msg_delivered);
    //test xp..
    rd_kafka_conf_set(conf, "debug", "all", test_errstr, sizeof(test_errstr));

    char err[100];
    rd_kafka_t *rk = rd_kafka_new(RD_KAFKA_PRODUCER, conf, errstr, sizeof(errstr));
    me->kafka_handler = rk;

   if (rk == NULL) {
            set_error("rd_kafka_new fail");
            return -1;
    }

    /* Add brokers */
    if (rd_kafka_brokers_add(me->kafka_handler, kafka_brokers) == 0) {
            fprintf(stderr, "%% No valid brokers specified\n");
            return -1;
    }
    return 0;

}

edenhill · 2014-01-14T10:29:19Z

Thanks for the extended debug info.
What version of librdkafka are you using? latest master?
There was a bug in 0.8.1 where some metadata requests would be badly formatted and cause an exception on the broker.

It was fixed in 16915a2

In any case using the latest master is favourable.

xianping · 2014-01-14T11:20:57Z

we use kafka_2.8.0-0.8.0 and librdkafka 0.8 version. which version should be ok?

xianping · 2014-01-14T11:31:41Z

when using master code, after 'make' ,there is no *.so in current path. Errors in makefile?

edenhill · 2014-01-14T12:24:17Z

I would like you to try the master branch.

Uhm, there should be .so files built.
What platform are you building on?
Can you provide the output of 'make clean all' ?

xianping · 2014-01-15T01:52:32Z

CentOS release 6.3 (Final)
-rw-r--r--. 1 root root 723784 Jan 15 09:49 librdkafka.a
-rw-r--r--. 1 root root 83344 Jan 15 09:49 snappy.o
-rw-r--r--. 1 root root 2849 Jan 15 09:49 snappy.d
-rw-r--r--. 1 root root 12592 Jan 15 09:49 rdlog.o
-rw-r--r--. 1 root root 3931 Jan 15 09:49 rdlog.d
-rw-r--r--. 1 root root 18680 Jan 15 09:49 rdqueue.o
-rw-r--r--. 1 root root 3873 Jan 15 09:49 rdqueue.d
-rw-r--r--. 1 root root 22400 Jan 15 09:49 rdthread.o
-rw-r--r--. 1 root root 4026 Jan 15 09:49 rdthread.d
-rw-r--r--. 1 root root 7736 Jan 15 09:49 rdrand.o
-rw-r--r--. 1 root root 3243 Jan 15 09:49 rdrand.d
-rw-r--r--. 1 root root 19288 Jan 15 09:49 rdaddr.o
-rw-r--r--. 1 root root 4105 Jan 15 09:49 rdaddr.d
-rw-r--r--. 1 root root 11272 Jan 15 09:49 rdgz.o
-rw-r--r--. 1 root root 3321 Jan 15 09:49 rdgz.d
-rw-r--r--. 1 root root 6560 Jan 15 09:49 rdcrc32.o
-rw-r--r--. 1 root root 1501 Jan 15 09:49 rdcrc32.d
-rw-r--r--. 1 root root 63656 Jan 15 09:49 rdkafka_defaultconf.o
-rw-r--r--. 1 root root 4820 Jan 15 09:49 rdkafka_defaultconf.d
-rw-r--r--. 1 root root 81232 Jan 15 09:49 rdkafka_topic.o
-rw-r--r--. 1 root root 4881 Jan 15 09:49 rdkafka_topic.d
-rw-r--r--. 1 root root 44280 Jan 15 09:49 rdkafka_msg.o
-rw-r--r--. 1 root root 4861 Jan 15 09:49 rdkafka_msg.d
-rw-r--r--. 1 root root 234704 Jan 15 09:49 rdkafka_broker.o
-rw-r--r--. 1 root root 5757 Jan 15 09:49 rdkafka_broker.d
-rw-r--r--. 1 root root 114056 Jan 15 09:49 rdkafka.o
-rw-r--r--. 1 root root 4933 Jan 15 09:49 rdkafka.d

edenhill · 2014-01-15T02:24:34Z

Thanks.
Can you provide the output from 'uname -s' ?

And also preferably the output from: 'make clean all'

edenhill · 2014-01-15T02:54:40Z

Issue #61 regarding .so generation is now fixed

xianping · 2014-01-15T08:40:34Z

Thanks for your patient, it can work now by using master branch.
One more question, in above 'mercury_sendmsg_kafka' function code, it will be called once for each message arrival. I think creating and destroying topic handler for each message is a heavy way. Because we have limit topic number, any way we can reuse the topic handler? I think this will be more effective, right?

Thanks

edenhill · 2014-01-15T09:47:00Z

You should typically hold on to the rkt (rd_kafka_topic_t *) and reuse it for producing your messages, so I would store it with the rk pointer (rd_kafka_t *) in your own struct and then reuse it as necessary.

xianping closed this as completed Jan 24, 2014

MockingJayWong mentioned this issue Jul 9, 2019

rd_kafka_topic_new after using admin api #2399

Closed

7 tasks

Furuta-Masakazu-quick mentioned this issue Sep 29, 2020

Producer detects "All broker connections are down" when rolling restarts #3090

Closed

7 tasks

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

nginx worker processes exit with signal 11 when publishing messages in nginx #56

nginx worker processes exit with signal 11 when publishing messages in nginx #56

xianping commented Jan 9, 2014

edenhill commented Jan 9, 2014

xianping commented Jan 9, 2014

edenhill commented Jan 9, 2014

xianping commented Jan 10, 2014

xianping commented Jan 10, 2014

edenhill commented Jan 10, 2014

xianping commented Jan 10, 2014

xianping commented Jan 13, 2014

edenhill commented Jan 13, 2014

edenhill commented Jan 13, 2014

xianping commented Jan 14, 2014

xianping commented Jan 14, 2014

edenhill commented Jan 14, 2014

xianping commented Jan 14, 2014

xianping commented Jan 14, 2014

edenhill commented Jan 14, 2014

xianping commented Jan 15, 2014

edenhill commented Jan 15, 2014

edenhill commented Jan 15, 2014

xianping commented Jan 15, 2014

edenhill commented Jan 15, 2014

nginx worker processes exit with signal 11 when publishing messages in nginx #56

nginx worker processes exit with signal 11 when publishing messages in nginx #56

Comments

xianping commented Jan 9, 2014

edenhill commented Jan 9, 2014

xianping commented Jan 9, 2014

edenhill commented Jan 9, 2014

xianping commented Jan 10, 2014

xianping commented Jan 10, 2014

edenhill commented Jan 10, 2014

xianping commented Jan 10, 2014

xianping commented Jan 13, 2014

edenhill commented Jan 13, 2014

edenhill commented Jan 13, 2014

xianping commented Jan 14, 2014

xianping commented Jan 14, 2014

edenhill commented Jan 14, 2014

xianping commented Jan 14, 2014

xianping commented Jan 14, 2014

edenhill commented Jan 14, 2014

xianping commented Jan 15, 2014

edenhill commented Jan 15, 2014

edenhill commented Jan 15, 2014

xianping commented Jan 15, 2014

edenhill commented Jan 15, 2014