search

pipelinedb / pipeline_kafka

PipelineDB extension for Kafka support
Other
61 stars 22 forks source link

LOG: worker process: worker0 [pipeline] (PID 16887) was terminated by signal 11: Segmentation fault #72

Open gb198871 opened 7 years ago

gb198871 commented 7 years ago

hi, I only use pipeline_kafka.consume_begin is no propelm, but use pipeline_kafka.emit_tuple have this propelm.

CREATE STREAM kafka_userlog_stream (time_tamp bigint, uuid text, age bigint,a1 text,a2 bigint,a3 numeric,a4 bool);

CREATE CONTINUOUS VIEW LOG_COUNT_VIEW WITH (sw = '30 minute') as select count(*) from kafka_userlog_stream;

CREATE TABLE DIM_AGE(begin_age int,end_age int, catalog varchar(300),PRIMARY KEY(begin_age,end_age)); insert into DIM_AGE values(0,30,'少年'); insert into DIM_AGE values(30,60,'中年'); insert into DIM_AGE values(60,120,'老年');

CREATE CONTINUOUS TRANSFORM CT_USER_AGE_CATALOG_TOKAFKA AS SELECT s.time_tamp::bigint, s.uuid::text,a.begin_age::int,s.a2::text,a.catalog::text FROM kafka_userlog_stream s JOIN DIM_AGE a ON s.age >= a.begin_age and s.age >= a.end_age THEN EXECUTE PROCEDURE pipeline_kafka.emit_tuple('pipelinedbTriggerTest');

SELECT pipeline_kafka.consume_begin('pipelineUserlog', 'kafka_userlog_stream',format := 'text', delimiter := E'|',batchsize := 1000, maxbytes := 32000000, parallelism := 5,start_offset := '-1');

error log：

LOG: worker process: worker0 [pipeline] (PID 16887) was terminated by signal 11: Segmentation fault DETAIL: Failed process was running: worker0 [pipeline] LOG: terminating any other active server processes WARNING: terminating connection because of crash of another server process DETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. HINT: In a moment you should be able to reconnect to the database and repeat your command. WARNING: terminating connection because of crash of another server process DETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. HINT: In a moment you should be able to reconnect to the database and repeat your command. WARNING: terminating connection because of crash of another server process DETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. HINT: In a moment you should be able to reconnect to the database and repeat your command. LOG: all server processes terminated; reinitializing LOG: database system was interrupted; last known up at 2017-11-06 18:14:24 CST LOG: database system was not properly shut down; automatic recovery in progress LOG: redo starts at 18/44E7DEE8 LOG: unexpected pageaddr 17/FA5CC000 in log segment 000000010000001800000045, offset 6078464 LOG: redo done at 18/455CBB68 LOG: last completed transaction was at log time 2017-11-06 18:16:55.555403+08 LOG: MultiXact member wraparound protections are now enabled

derekjn commented 7 years ago

@gb198871 thanks for the report! Is there no stack trace in the logs?

gb198871 commented 7 years ago

@derekjn All are build by master pipelinedb\ pipeline_kafka\ librdkafka ，how can i config print stack trace logs ? thanks!!!

gb198871 commented 7 years ago

2017-11-18 17:26:59 CSTLOG: 00000: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 4448): librdkafka error: [thrd:slave-67:9090/bootstrap]: slave-67:9090/3: Receive failed: Disconnected 2017-11-18 17:26:59 CSTLOCATION: consume_topic_into_relation, pipeline_kafka.c:1008 PID: 4448 2017-11-18 17:26:59 CSTLOG: 00000: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 4446): librdkafka error: [thrd:slave-75:9090/bootstrap]: slave-75:9090/2: Receive failed: Disconnected 2017-11-18 17:26:59 CSTLOCATION: consume_topic_into_relation, pipeline_kafka.c:1008 PID: 4446 2017-11-18 17:26:59 CSTLOG: 00000: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 4450): librdkafka error: [thrd:slave-75:9090/bootstrap]: slave-75:9090/2: Receive failed: Disconnected 2017-11-18 17:26:59 CSTLOCATION: consume_topic_into_relation, pipeline_kafka.c:1008 PID: 4450 2017-11-18 17:26:59 CSTLOG: 00000: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 4447): librdkafka error: [thrd:slave-63:9090/bootstrap]: slave-63:9090/1: Receive failed: Disconnected 2017-11-18 17:26:59 CSTLOCATION: consume_topic_into_relation, pipeline_kafka.c:1008 PID: 4447 2017-11-18 17:26:59 CSTLOG: 00000: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 4449): librdkafka error: [thrd:slave-63:9090/bootstrap]: slave-63:9090/1: Receive failed: Disconnected 2017-11-18 17:26:59 CSTLOCATION: consume_topic_into_relation, pipeline_kafka.c:1008 PID: 4449 2017-11-18 17:26:59 CSTLOG: 00000: [pipeline_kafka producer]: [thrd:slave-63:9090/bootstrap]: slave-63:9090/1: Receive failed: Disconnected 2017-11-18 17:26:59 CSTLOCATION: kafka_produce_msg, pipeline_kafka.c:2128 PID: 4423 2017-11-18 17:26:59 CSTSTATEMENT: ct_user_age_catalog_tokafka2 2017-11-18 17:26:59 CSTLOG: 00000: worker process: worker0 [pipeline] (PID 4424) was terminated by signal 11: Segmentation fault 2017-11-18 17:26:59 CSTDETAIL: Failed process was running: worker0 [pipeline] 2017-11-18 17:26:59 CSTLOCATION: LogChildExit, postmaster.c:3534 PID: 4408 2017-11-18 17:26:59 CSTLOG: 00000: terminating any other active server processes 2017-11-18 17:26:59 CSTLOCATION: HandleChildCrash, postmaster.c:3231 PID: 4408 2017-11-18 17:26:59 CSTWARNING: 57P02: terminating connection because of crash of another server process 2017-11-18 17:26:59 CSTDETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. 2017-11-18 17:26:59 CSTHINT: In a moment you should be able to reconnect to the database and repeat your command. 2017-11-18 17:26:59 CSTLOCATION: quickdie, postgres.c:2653 PID: 4416 2017-11-18 17:26:59 CSTWARNING: 57P02: terminating connection because of crash of another server process 2017-11-18 17:26:59 CSTDETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. 2017-11-18 17:26:59 CSTHINT: In a moment you should be able to reconnect to the database and repeat your command. 2017-11-18 17:26:59 CSTLOCATION: quickdie, postgres.c:2653 PID: 4414 2017-11-18 17:26:59 CSTWARNING: 57P02: terminating connection because of crash of another server process 2017-11-18 17:26:59 CSTDETAIL: The postmaster has commanded this server process to roll back the current transaction and exit, because another server process exited abnormally and possibly corrupted shared memory. 2017-11-18 17:26:59 CSTHINT: In a moment you should be able to reconnect to the database and repeat your command. 2017-11-18 17:26:59 CSTLOCATION: quickdie, postgres.c:2653 PID: 4445 Error in `pipelinedb: bgworker: worker1 [pipeline] ': munmap_chunk(): invalid pointer: 0x00000000010ee0e0 ======= Backtrace: ========= /lib64/libc.so.6(+0x7ab54)[0x7f8e873c7b54] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_msg_destroy+0x79)[0x7f8e80b77cb9] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_dr_msgq+0x6f)[0x7f8e80b705cf] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x57db5)[0x7f8e80b91db5] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_buf_callback+0x4d)[0x7f8e80b8659d] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_recv+0x3e7)[0x7f8e80b6e117] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x4a0b8)[0x7f8e80b840b8] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x3b0c8)[0x7f8e80b750c8] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x3bfd5)[0x7f8e80b75fd5] /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x83f87)[0x7f8e80bbdf87] /lib64/libpthread.so.0(+0x7e25)[0x7f8e88343e25] /lib64/libc.so.6(clone+0x6d)[0x7f8e8744534d] ======= Memory map: ======== 00400000-00a65000 r-xp 00000000 fd:00 36201372 /usr/local/pipelinedb/bin/pipelinedb 00c64000-00c65000 r--p 00664000 fd:00 36201372 /usr/local/pipelinedb/bin/pipelinedb 00c65000-00c73000 rw-p 00665000 fd:00 36201372 /usr/local/pipelinedb/bin/pipelinedb 00c73000-00d3c000 rw-p 00000000 00:00 0 00f47000-00f85000 rw-p 00000000 00:00 0 [heap] 00f85000-03111000 rw-p 00000000 00:00 0 [heap] 7f8c54000000-7f8c54022000 rw-p 00000000 00:00 0 7f8c54022000-7f8c58000000 ---p 00000000 00:00 0 7f8c58000000-7f8c587d9000 rw-p 00000000 00:00 0 7f8c587d9000-7f8c5c000000 ---p 00000000 00:00 0 7f8c5c000000-7f8c5c8e9000 rw-p 00000000 00:00 0 7f8c5c8e9000-7f8c60000000 ---p 00000000 00:00 0 7f8c627fd000-7f8c627fe000 ---p 00000000 00:00 0 7f8c627fe000-7f8c62ffe000 rw-p 00000000 00:00 0 [stack:4488] 7f8c62ffe000-7f8c62fff000 ---p 00000000 00:00 0 7f8c62fff000-7f8c637ff000 rw-p 00000000 00:00 0 [stack:4487] 7f8c637ff000-7f8c63800000 ---p 00000000 00:00 0 7f8c63800000-7f8c64000000 rw-p 00000000 00:00 0 [stack:4486] 7f8c64000000-7f8c64022000 rw-p 00000000 00:00 0 7f8c64022000-7f8c68000000 ---p 00000000 00:00 0 7f8c68000000-7f8c68022000 rw-p 00000000 00:00 0 7f8c68022000-7f8c6c000000 ---p 00000000 00:00 0 7f8c6c000000-7f8c6c0d3000 rw-p 00000000 00:00 0 7f8c6c0d3000-7f8c70000000 ---p 00000000 00:00 0 7f8c702b9000-7f8c702ba000 ---p 00000000 00:00 0 7f8c702ba000-7f8c70aba000 rw-p 00000000 00:00 0 [stack:4485] 7f8c70aba000-7f8c70abb000 ---p 00000000 00:00 0 7f8c70abb000-7f8c712bb000 rw-p 00000000 00:00 0 [stack:4484] 7f8c712bb000-7f8c712bc000 ---p 00000000 00:00 0 7f8c712bc000-7f8c71abc000 rw-p 00000000 00:00 0 [stack:4440] 7f8c71abc000-7f8c71abd000 ---p 00000000 00:00 0 7f8c71abd000-7f8c722bd000 rw-p 00000000 00:00 0 [stack:4439] 7f8c722bd000-7f8c722c9000 r-xp 00000000 fd:00 68876664 /usr/lib64/libnss_files-2.17.so 7f8c722c9000-7f8c724c8000 ---p 0000c000 fd:00 68876664 /usr/lib64/libnss_files-2.17.so 7f8c724c8000-7f8c724c9000 r--p 0000b000 fd:00 68876664 /usr/lib64/libnss_files-2.17.so 7f8c724c9000-7f8c724ca000 rw-p 0000c000 fd:00 68876664 /usr/lib64/libnss_files-2.17.so 7f8c724ca000-7f8c724d0000 rw-p 00000000 00:00 0 7f8c724d0000-7f8e80924000 rw-s 00000000 00:04 176899 /dev/zero (deleted) 7f8e80924000-7f8e80939000 r-xp 00000000 fd:00 69423250 /usr/lib64/libz.so.1.2.7 7f8e80939000-7f8e80b38000 ---p 00015000 fd:00 69423250 /usr/lib64/libz.so.1.2.7 7f8e80b38000-7f8e80b39000 r--p 00014000 fd:00 69423250 /usr/lib64/libz.so.1.2.7 7f8e80b39000-7f8e80b3a000 rw-p 00015000 fd:00 69423250 /usr/lib64/libz.so.1.2.7 7f8e80b3a000-7f8e80c11000 r-xp 00000000 fd:00 101990311 /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so 7f8e80c11000-7f8e80e11000 ---p 000d7000 fd:00 101990311 /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so 7f8e80e11000-7f8e80e1e000 r--p 000d7000 fd:00 101990311 /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so 7f8e80e1e000-7f8e80e20000 rw-p 000e4000 fd:00 101990311 /usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so 7f8e80e20000-7f8e80e24000 rw-p 00000000 00:00 0 7f8e80e24000-7f8e8734d000 r--p 00000000 fd:00 68876655 /usr/lib/locale/locale-archive 7f8e8734d000-7f8e87505000 r-xp 00000000 fd:00 67267087 /usr/lib64/libc-2.17.so 7f8e87505000-7f8e87705000 ---p 001b8000 fd:00 67267087 /usr/lib64/libc-2.17.so 7f8e87705000-7f8e87709000 r--p 001b8000 fd:00 67267087 /usr/lib64/libc-2.17.so 7f8e87709000-7f8e8770b000 rw-p 001bc000 fd:00 67267087 /usr/lib64/libc-2.17.so 7f8e8770b000-7f8e87710000 rw-p 00000000 00:00 0 7f8e87710000-7f8e87725000 r-xp 00000000 fd:00 68876648 /usr/lib64/libgcc_s-4.8.5-20150702.so.1 7f8e87725000-7f8e87924000 ---p 00015000 fd:00 68876648 /usr/lib64/libgcc_s-4.8.5-20150702.so.1 7f8e87924000-7f8e87925000 r--p 00014000 fd:00 68876648 /usr/lib64/libgcc_s-4.8.5-20150702.so.1 7f8e87925000-7f8e87926000 rw-p 00015000 fd:00 68876648 /usr/lib64/libgcc_s-4.8.5-20150702.so.1 7f8e87926000-7f8e87a27000 r-xp 00000000 fd:00 67267097 /usr/lib64/libm-2.17.so 7f8e87a27000-7f8e87c26000 ---p 00101000 fd:00 67267097 /usr/lib64/libm-2.17.so 7f8e87c26000-7f8e87c27000 r--p 00100000 fd:00 67267097 /usr/lib64/libm-2.17.so 7f8e87c27000-7f8e87c28000 rw-p 00101000 fd:00 67267097 /usr/lib64/libm-2.17.so 7f8e87c28000-7f8e87c2a000 r-xp 00000000 fd:00 67267094 /usr/lib64/libdl-2.17.so 7f8e87c2a000-7f8e87e2a000 ---p 00002000 fd:00 67267094 /usr/lib64/libdl-2.17.so 7f8e87e2a000-7f8e87e2b000 r--p 00002000 fd:00 67267094 /usr/lib64/libdl-2.17.so 7f8e87e2b000-7f8e87e2c000 rw-p 00003000 fd:00 67267094 /usr/lib64/libdl-2.17.so 7f8e87e2c000-7f8e87e33000 r-xp 00000000 fd:00 67267106 /usr/lib64/librt-2.17.so 7f8e87e33000-7f8e88032000 ---p 00007000 fd:00 67267106 /usr/lib64/librt-2.17.so 7f8e88032000-7f8e88033000 r--p 00006000 fd:00 67267106 /usr/lib64/librt-2.17.so 7f8e88033000-7f8e88034000 rw-p 00007000 fd:00 67267106 /usr/lib64/librt-2.17.so 7f8e88034000-7f8e8811d000 r-xp 00000000 fd:00 67267744 /usr/lib64/libstdc++.so.6.0.19 7f8e8811d000-7f8e8831d000 ---p 000e9000 fd:00 67267744 /usr/lib64/libstdc++.so.6.0.19 7f8e8831d000-7f8e88325000 r--p 000e9000 fd:00 67267744 /usr/lib64/libstdc++.so.6.0.19 7f8e88325000-7f8e88327000 rw-p 000f1000 fd:00 67267744 /usr/lib64/libstdc++.so.6.0.19 7f8e88327000-7f8e8833c000 rw-p 00000000 00:00 0 7f8e8833c000-7f8e88353000 r-xp 00000000 fd:00 68876671 /usr/lib64/libpthread-2.17.so 7f8e88353000-7f8e88552000 ---p 00017000 fd:00 68876671 /usr/lib64/libpthread-2.17.so 7f8e88552000-7f8e88553000 r--p 00016000 fd:00 68876671 /usr/lib64/libpthread-2.17.so 7f8e88553000-7f8e88554000 rw-p 00017000 fd:00 68876671 /usr/lib64/libpthread-2.17.so 7f8e88554000-7f8e88558000 rw-p 00000000 00:00 0 7f8e88558000-7f8e88579000 r-xp 00000000 fd:00 68876654 /usr/lib64/ld-2.17.so 7f8e885c6000-7f8e88647000 rw-p 00000000 00:00 0 7f8e88709000-7f8e8876b000 rw-p 00000000 00:00 0 7f8e8876b000-7f8e88772000 rw-p 00000000 00:00 0 7f8e88774000-7f8e88775000 rw-p 00000000 00:00 0 7f8e88775000-7f8e88776000 rw-p 00000000 00:00 0 7f8e88776000-7f8e88777000 rw-s 00000000 00:12 176901 /dev/shm/PostgreSQL.1804289383 7f8e88777000-7f8e88778000 rw-s 00000000 00:04 163840 /SYSV0052e2c1 (deleted) 7f8e88778000-7f8e88779000 rw-p 00000000 00:00 0 7f8e88779000-7f8e8877a000 r--p 00021000 fd:00 68876654 /usr/lib64/ld-2.17.so 7f8e8877a000-7f8e8877b000 rw-p 00022000 fd:00 68876654 /usr/lib64/ld-2.17.so 7f8e8877b000-7f8e8877c000 rw-p 00000000 00:00 0 7fff504ec000-7fff50511000 rw-p 00000000 00:00 0 [stack] 7fff50558000-7fff5055a000 r-xp 00000000 00:00 0 [vdso] ffffffffff600000-ffffffffff601000 r-xp 00000000 00:00 0 [vsyscall]

derekjn commented 7 years ago

Relevant stack trace:

/lib64/libc.so.6(+0x7ab54)[0x7f8e873c7b54]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_msg_destroy+0x79)[0x7f8e80b77cb9]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_dr_msgq+0x6f)[0x7f8e80b705cf]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x57db5)[0x7f8e80b91db5]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_buf_callback+0x4d)[0x7f8e80b8659d]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(rd_kafka_recv+0x3e7)[0x7f8e80b6e117]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x4a0b8)[0x7f8e80b840b8]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x3b0c8)[0x7f8e80b750c8]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x3bfd5)[0x7f8e80b75fd5]
/usr/local/pipelinedb/lib/pipelinedb/pipeline_kafka.so(+0x83f87)[0x7f8e80bbdf87]

@gb198871 thanks, this is very helpful!

gb198871 commented 7 years ago

@derekjn this is a bug or configuration problems ？ can repair ？thanks！

derekjn commented 7 years ago

@gb198871 are you able to provide any sample data to reproduce this issue with? If not we'll keep digging but that would make it really easy for us.

gb198871 commented 7 years ago

I am try to debug code,find out throw exception row,but i don't know how to repair: `void rd_kafka_msg_destroy (rd_kafka_t rk, rd_kafka_msg_t rkm) {

if (rkm->rkm_flags & RD_KAFKA_MSG_F_ACCOUNT) {
    rd_dassert(rk || rkm->rkm_rkmessage.rkt);
    rd_kafka_curr_msgs_sub(
        rk ? rk :
        rd_kafka_topic_a2i(rkm->rkm_rkmessage.rkt)->rkt_rk,
        1, rkm->rkm_len);
}

if (likely(rkm->rkm_rkmessage.rkt != NULL))
    rd_kafka_topic_destroy0(
                    rd_kafka_topic_a2s(rkm->rkm_rkmessage.rkt));

if (rkm->rkm_flags & RD_KAFKA_MSG_F_FREE && rkm->rkm_payload)
    rd_free(rkm->rkm_payload);

if (rkm->rkm_flags & RD_KAFKA_MSG_F_FREE_RKM)
    rd_free(rkm); // this row throw exception

}`

sample data produce to kafka topic "pipelineUserlog" 40000 record one second： 1511918562082|ae9684d9-dd81-47a3-8492-cfa559ecdf5e|27|f7e3482c-d37e-4221-932d-14e3a41d1f9e|-5040975481107957024|0.2535811422823496|false 1511918562082|da372c54-ff31-41fa-9860-e6341fd18036|9|3d677a68-bce6-448f-bb5c-2880e30db3f7|-1041782078399413188|0.039479525953184935|true 1511918562082|7bee96ab-6712-4572-8e0e-3b6b66e927c9|22|e33fa084-fe0e-4ebb-bcd1-6d4d68f48761|4982898864022778410|0.644946324772111|true

I discovery when add config on pipeline_kafka.c kafka_produce_msg function recompile: rd_kafka_conf_set(conf,"batch.num.messages","1000000",err_msg, sizeof(err_msg)); rd_kafka_conf_set(conf,"queue.buffering.max.messages","10000000",err_msg, sizeof(err_msg)); rd_kafka_conf_set(conf,"message.max.bytes","100000000",err_msg, sizeof(err_msg));

and comsumer kafka data batchSize change smaller 500: SELECT pipeline_kafka.consume_begin('pipelineUserlog', 'kafka_userlog_stream',format := 'text', delimiter := E'|',batchsize := 500, maxbytes := 32000000, parallelism := 5,start_offset := '-1');

It is running more than 24 hour ,did not happend ' was terminated by signal 11: Segmentation fault ' error.

Another how to configuration optimization pipeinedb.conf ?

thank you very much!

gb198871 commented 7 years ago

I am try debug code ,find throw exception on this row,but i don’t know how to repairkkk2017-11-29 09:09:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9359): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:14:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9360): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected 2017-11-29 09:14:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9358): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected 2017-11-29 09:14:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9361): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:14:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9360): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:14:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9357): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:14:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9358): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:19:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9359): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected 2017-11-29 09:24:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9360): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected 2017-11-29 09:24:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9357): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:24:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9361): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected 2017-11-29 09:29:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9359): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:29:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9358): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected 2017-11-29 09:34:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9361): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:34:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9360): librdkafka error: [thrd:slave-75:9090/2]: slave-75:9090/2: Receive failed: Disconnected 2017-11-29 09:39:18 CSTLOG: [pipeline_kafka] kafka_userlog_stream <- pipelineUserlog (PID 9357): librdkafka error: [thrd:slave-67:9090/3]: slave-67:9090/3: Receive failed: Disconnected

Always print this error but running normal still