mirror of
https://github.com/rsyslog/rsyslog.git
synced 2025-12-19 04:30:43 +01:00
Imkafka: Added multithreaded support for kafka consumers (#3013)
* imkafka: implement multithreading support for kafka consumers. Each consumer runs in it's own consumer thread now. New tests have also been added for this.
This commit is contained in:
Rainer Gerhards
parent
8f12199cc4
commit
97308b5610
@ -61,6 +61,10 @@ DEFobjCurrIf(ruleset)
|
||||
DEFobjCurrIf(glbl)
|
||||
DEFobjCurrIf(statsobj)
|
||||
|
||||
/* forward references */
|
||||
static void * imkafkawrkr(void *myself);
|
||||
|
||||
|
||||
struct kafka_params {
|
||||
const char *name;
|
||||
const char *val;
|
||||
@ -108,6 +112,17 @@ struct modConfData_s {
|
||||
uchar *pszBindRuleset; /* default name of Ruleset to bind to */
|
||||
};
|
||||
|
||||
/* global data */
|
||||
pthread_attr_t wrkrThrdAttr; /* Attribute for worker threads ; read only after startup */
|
||||
static int activeKafkaworkers = 0;
|
||||
/* The following structure controls the worker threads. Global data is
|
||||
* needed for their access.
|
||||
*/
|
||||
static struct kafkaWrkrInfo_s {
|
||||
pthread_t tid; /* the worker's thread ID */
|
||||
instanceConf_t *inst; /* Pointer to imkafka instance */
|
||||
} *kafkaWrkrInfo;
|
||||
|
||||
static modConfData_t *loadModConf = NULL;/* modConf ptr to use for the current load process */
|
||||
static modConfData_t *runModConf = NULL;/* modConf ptr to use for the current load process */
|
||||
|
||||
@ -692,17 +707,18 @@ ENDfreeCnf
|
||||
/* This function is called to gather input.
|
||||
*/
|
||||
BEGINrunInput
|
||||
int i;
|
||||
instanceConf_t *inst;
|
||||
CODESTARTrunInput
|
||||
DBGPRINTF("imkafka: runInput loop started ...\n");
|
||||
int activeListeners = 0;
|
||||
activeKafkaworkers = 0;
|
||||
for(inst = runModConf->root ; inst != NULL ; inst = inst->next) {
|
||||
if(inst->rk != NULL) {
|
||||
++activeListeners;
|
||||
++activeKafkaworkers;
|
||||
}
|
||||
}
|
||||
|
||||
if(activeListeners == 0) {
|
||||
if(activeKafkaworkers == 0) {
|
||||
LogError(0, RS_RET_ERR, "imkafka: no active inputs, input does "
|
||||
"not run - there should have been additional error "
|
||||
"messages given previously");
|
||||
@ -710,36 +726,32 @@ CODESTARTrunInput
|
||||
}
|
||||
|
||||
|
||||
/* Start endless consumer loop - it is terminated when the thread is
|
||||
* signalled to do so. This, however, is handled by the framework.
|
||||
DBGPRINTF("imkafka: Starting %d imkafka workerthreads\n", activeKafkaworkers);
|
||||
kafkaWrkrInfo = calloc(activeKafkaworkers, sizeof(struct kafkaWrkrInfo_s));
|
||||
if (kafkaWrkrInfo == NULL) {
|
||||
LogError(errno, RS_RET_OUT_OF_MEMORY, "imkafka: worker-info array allocation failed.");
|
||||
ABORT_FINALIZE(RS_RET_OUT_OF_MEMORY);
|
||||
}
|
||||
|
||||
/* Start worker threads for each imkafka input source
|
||||
*/
|
||||
do {
|
||||
i = 0;
|
||||
for(inst = runModConf->root ; inst != NULL ; inst = inst->next) {
|
||||
if(glbl.GetGlobalInputTermState() == 1)
|
||||
break; /* terminate input! */
|
||||
|
||||
if(inst->rk == NULL) {
|
||||
continue;
|
||||
/* init worker info structure! */
|
||||
kafkaWrkrInfo[i].inst = inst; /* Set reference pointer */
|
||||
pthread_create(&kafkaWrkrInfo[i].tid, &wrkrThrdAttr, imkafkawrkr, &(kafkaWrkrInfo[i]));
|
||||
i++;
|
||||
}
|
||||
|
||||
// Try to add consumer only if connected! */
|
||||
if(inst->bIsConnected == 1 && inst->bIsSubscribed == 0 ) {
|
||||
addConsumer(runModConf, inst);
|
||||
}
|
||||
if(inst->bIsSubscribed == 1 ) {
|
||||
msgConsume(inst);
|
||||
}
|
||||
}
|
||||
while(glbl.GetGlobalInputTermState() == 0) {
|
||||
|
||||
/* Note: the additional 10000ns wait is vitally important. It guards rsyslog
|
||||
* against totally hogging the CPU if the users selects a polling interval
|
||||
* of 0 seconds. It doesn't hurt any other valid scenario. So do not remove.
|
||||
* rgerhards, 2008-02-14
|
||||
*/
|
||||
if(glbl.GetGlobalInputTermState() == 0)
|
||||
srSleep(0, 100000);
|
||||
} while(glbl.GetGlobalInputTermState() == 0);
|
||||
|
||||
}
|
||||
DBGPRINTF("imkafka: terminating upon request of rsyslog core\n");
|
||||
finalize_it:
|
||||
ENDrunInput
|
||||
@ -758,6 +770,15 @@ ENDwillRun
|
||||
|
||||
BEGINafterRun
|
||||
CODESTARTafterRun
|
||||
/* Cleanup imkafka worker threads */
|
||||
int i;
|
||||
DBGPRINTF("imkafka: Stopping imkafka workerthreads\n");
|
||||
for(i = 0 ; i < activeKafkaworkers ; ++i) {
|
||||
pthread_join(kafkaWrkrInfo[i].tid, NULL);
|
||||
DBGPRINTF("imkafka: Stopped worker %d\n", i);
|
||||
}
|
||||
free(kafkaWrkrInfo);
|
||||
|
||||
/* do cleanup here */
|
||||
if(pInputName != NULL)
|
||||
prop.Destruct(&pInputName);
|
||||
@ -793,6 +814,7 @@ ENDafterRun
|
||||
|
||||
BEGINmodExit
|
||||
CODESTARTmodExit
|
||||
pthread_attr_destroy(&wrkrThrdAttr);
|
||||
/* release objects we used */
|
||||
objRelease(statsobj, CORE_COMPONENT);
|
||||
objRelease(ruleset, CORE_COMPONENT);
|
||||
@ -828,7 +850,50 @@ CODEmodInit_QueryRegCFSLineHdlr
|
||||
CHKiRet(objUse(prop, CORE_COMPONENT));
|
||||
CHKiRet(objUse(ruleset, CORE_COMPONENT));
|
||||
CHKiRet(objUse(statsobj, CORE_COMPONENT));
|
||||
|
||||
/* initialize "read-only" thread attributes */
|
||||
pthread_attr_init(&wrkrThrdAttr);
|
||||
pthread_attr_setstacksize(&wrkrThrdAttr, 4096*1024);
|
||||
|
||||
DBGPRINTF("imkafka %s using librdkafka version %s, 0x%x\n",
|
||||
VERSION, rd_kafka_version_str(), rd_kafka_version());
|
||||
|
||||
ENDmodInit
|
||||
|
||||
/*
|
||||
* Workerthread function for a single kafka consomer
|
||||
*/
|
||||
static void *
|
||||
imkafkawrkr(void *myself)
|
||||
{
|
||||
struct kafkaWrkrInfo_s *me = (struct kafkaWrkrInfo_s*) myself;
|
||||
DBGPRINTF("imkafka: started kafka consumer workerthread on %s/%s/%s\n",
|
||||
me->inst->topic, me->inst->consumergroup, me->inst->brokers);
|
||||
|
||||
do {
|
||||
if(glbl.GetGlobalInputTermState() == 1)
|
||||
break; /* terminate input! */
|
||||
|
||||
if(me->inst->rk == NULL) {
|
||||
continue;
|
||||
}
|
||||
|
||||
// Try to add consumer only if connected! */
|
||||
if(me->inst->bIsConnected == 1 && me->inst->bIsSubscribed == 0 ) {
|
||||
addConsumer(runModConf, me->inst);
|
||||
}
|
||||
if(me->inst->bIsSubscribed == 1 ) {
|
||||
msgConsume(me->inst);
|
||||
}
|
||||
/* Note: the additional 10000ns wait is vitally important. It guards rsyslog
|
||||
* against totally hogging the CPU if the users selects a polling interval
|
||||
* of 0 seconds. It doesn't hurt any other valid scenario. So do not remove.
|
||||
* rgerhards, 2008-02-14
|
||||
*/
|
||||
if(glbl.GetGlobalInputTermState() == 0)
|
||||
srSleep(0, 100000);
|
||||
} while(glbl.GetGlobalInputTermState() == 0);
|
||||
|
||||
DBGPRINTF("imkafka: stopped kafka consumer workerthread on %s/%s/%s\n",
|
||||
me->inst->topic, me->inst->consumergroup, me->inst->brokers);
|
||||
return NULL;
|
||||
}
|
||||
|
||||
@ -604,6 +604,7 @@ if ENABLE_KAFKA_TESTS
|
||||
TESTS += \
|
||||
omkafka.sh \
|
||||
imkafka.sh \
|
||||
imkafka_multi_single.sh \
|
||||
sndrcv_kafka.sh \
|
||||
sndrcv_kafka_multi_topics.sh
|
||||
# Tests below need to be stable first!
|
||||
@ -1624,6 +1625,7 @@ EXTRA_DIST= \
|
||||
mysql-actq-mt-withpause-vg.sh \
|
||||
omkafka.sh \
|
||||
imkafka.sh \
|
||||
imkafka_multi_single.sh \
|
||||
sndrcv_kafka.sh \
|
||||
sndrcv_kafka_multi_topics.sh \
|
||||
sndrcv_kafka-vg-rcvr.sh \
|
||||
|
||||
155
tests/imkafka_multi_many.sh
Executable file
155
tests/imkafka_multi_many.sh
Executable file
@ -0,0 +1,155 @@
|
||||
#!/bin/bash
|
||||
# added 2018-08-29 by alorbach
|
||||
# This file is part of the rsyslog project, released under ASL 2.0
|
||||
export TESTMESSAGES=100000
|
||||
export TESTMESSAGESFULL=100000
|
||||
|
||||
# Generate random topic name
|
||||
export RANDTOPIC=$(cat /dev/urandom | tr -dc 'a-zA-Z0-9' | fold -w 8 | head -n 1)
|
||||
|
||||
echo Init Testbench
|
||||
. $srcdir/diag.sh init
|
||||
|
||||
# Check for kafkacat
|
||||
check_command_available kafkacat
|
||||
|
||||
# enable the EXTRA_EXITCHECK only if really needed - otherwise spams the test log
|
||||
# too much
|
||||
#export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
echo ===============================================================================
|
||||
echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
. $srcdir/diag.sh download-kafka
|
||||
. $srcdir/diag.sh stop-zookeeper '.dep_wrk1'
|
||||
. $srcdir/diag.sh stop-zookeeper '.dep_wrk2'
|
||||
. $srcdir/diag.sh stop-zookeeper '.dep_wrk3'
|
||||
. $srcdir/diag.sh stop-kafka '.dep_wrk1'
|
||||
. $srcdir/diag.sh stop-kafka '.dep_wrk2'
|
||||
. $srcdir/diag.sh stop-kafka '.dep_wrk3'
|
||||
|
||||
echo Create kafka/zookeeper instance and topics
|
||||
. $srcdir/diag.sh start-zookeeper '.dep_wrk1'
|
||||
. $srcdir/diag.sh start-zookeeper '.dep_wrk2'
|
||||
. $srcdir/diag.sh start-zookeeper '.dep_wrk3'
|
||||
. $srcdir/diag.sh start-kafka '.dep_wrk1'
|
||||
. $srcdir/diag.sh start-kafka '.dep_wrk2'
|
||||
. $srcdir/diag.sh start-kafka '.dep_wrk3'
|
||||
|
||||
echo Give Kafka some time to sync...
|
||||
sleep 5
|
||||
|
||||
# create new topic
|
||||
. $srcdir/diag.sh create-kafka-topic $RANDTOPIC '.dep_wrk1' '22181'
|
||||
|
||||
echo Give Kafka some time to process topic create ...
|
||||
sleep 5
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
main_queue(queue.timeoutactioncompletion="60000" queue.timeoutshutdown="60000")
|
||||
|
||||
module(load="../plugins/imkafka/.libs/imkafka")
|
||||
/* Polls messages from kafka server!*/
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker=["localhost:29092", "localhost:29093", "localhost:29094"]
|
||||
consumergroup="default1"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker=["localhost:29092", "localhost:29093", "localhost:29094"]
|
||||
consumergroup="default2"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker=["localhost:29092", "localhost:29093", "localhost:29094"]
|
||||
consumergroup="default3"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker=["localhost:29092", "localhost:29093", "localhost:29094"]
|
||||
consumergroup="default4"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
template(name="outfmt" type="string" string="%msg:F,58:2%\n")
|
||||
|
||||
if ($msg contains "msgnum:") then {
|
||||
action( type="omfile" file=`echo $RSYSLOG_OUT_LOG` template="outfmt" )
|
||||
}
|
||||
'
|
||||
|
||||
# Start imkafka receiver config
|
||||
echo Starting receiver instance [imkafka]
|
||||
startup
|
||||
# ---
|
||||
|
||||
# Messure Starttime
|
||||
TIMESTART=$(date +%s.%N)
|
||||
|
||||
# --- Fill Kafka Server with messages
|
||||
# Can properly be done in a better way?!
|
||||
for i in {00000001..00100000}
|
||||
do
|
||||
echo " msgnum:$i" >> $RSYSLOG_OUT_LOG.in
|
||||
done
|
||||
|
||||
echo Inject messages into kafka
|
||||
cat $RSYSLOG_OUT_LOG.in | kafkacat -P -b localhost:29092 -t $RANDTOPIC
|
||||
# ---
|
||||
|
||||
echo Give imkafka some time to start...
|
||||
sleep 5
|
||||
|
||||
echo Stopping sender instance [omkafka]
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
# Messure Endtime
|
||||
TIMEEND=$(date +%s.%N)
|
||||
TIMEDIFF=$(echo "$TIMEEND - $TIMESTART" | bc)
|
||||
echo "*** imkafka time to process all data: $TIMEDIFF seconds!"
|
||||
|
||||
# Delete topic to remove old traces before
|
||||
. $srcdir/diag.sh delete-kafka-topic $RANDTOPIC '.dep_wrk1' '22181'
|
||||
|
||||
echo \[sndrcv_kafka.sh\]: stop kafka instances
|
||||
. $srcdir/diag.sh stop-kafka '.dep_wrk1'
|
||||
. $srcdir/diag.sh stop-kafka '.dep_wrk2'
|
||||
. $srcdir/diag.sh stop-kafka '.dep_wrk3'
|
||||
. $srcdir/diag.sh stop-zookeeper '.dep_wrk1'
|
||||
. $srcdir/diag.sh stop-zookeeper '.dep_wrk2'
|
||||
. $srcdir/diag.sh stop-zookeeper '.dep_wrk3'
|
||||
|
||||
# Do the final sequence check
|
||||
seq_check 1 $TESTMESSAGESFULL -d
|
||||
|
||||
echo success
|
||||
exit_test
|
||||
191
tests/imkafka_multi_single.sh
Executable file
191
tests/imkafka_multi_single.sh
Executable file
@ -0,0 +1,191 @@
|
||||
#!/bin/bash
|
||||
# added 2018-08-29 by alorbach
|
||||
# This file is part of the rsyslog project, released under ASL 2.0
|
||||
export TESTMESSAGES=100000
|
||||
export TESTMESSAGESFULL=100000
|
||||
|
||||
# Generate random topic name
|
||||
export RANDTOPIC=$(cat /dev/urandom | tr -dc 'a-zA-Z0-9' | fold -w 8 | head -n 1)
|
||||
|
||||
echo Init Testbench
|
||||
. $srcdir/diag.sh init
|
||||
|
||||
# Check for kafkacat
|
||||
check_command_available kafkacat
|
||||
|
||||
# enable the EXTRA_EXITCHECK only if really needed - otherwise spams the test log
|
||||
# too much
|
||||
#export EXTRA_EXITCHECK=dumpkafkalogs
|
||||
echo ===============================================================================
|
||||
echo Create kafka/zookeeper instance and $RANDTOPIC topic
|
||||
. $srcdir/diag.sh download-kafka
|
||||
. $srcdir/diag.sh stop-zookeeper
|
||||
. $srcdir/diag.sh stop-kafka
|
||||
|
||||
echo Create kafka/zookeeper instance and topics
|
||||
. $srcdir/diag.sh start-zookeeper
|
||||
. $srcdir/diag.sh start-kafka
|
||||
|
||||
echo Give Kafka some time to sync...
|
||||
sleep 5
|
||||
|
||||
# create new topic
|
||||
. $srcdir/diag.sh create-kafka-topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
echo Give Kafka some time to process topic create ...
|
||||
sleep 5
|
||||
|
||||
# --- Create imkafka receiver config
|
||||
export RSYSLOG_DEBUGLOG="log"
|
||||
generate_conf
|
||||
add_conf '
|
||||
main_queue(queue.timeoutactioncompletion="60000" queue.timeoutshutdown="60000")
|
||||
|
||||
module(load="../plugins/imkafka/.libs/imkafka")
|
||||
/* Polls messages from kafka server!*/
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default1"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default2"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default3"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default4"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default5"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default6"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default7"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
input( type="imkafka"
|
||||
topic="'$RANDTOPIC'"
|
||||
broker="localhost:29092"
|
||||
consumergroup="default8"
|
||||
confParam=[ "compression.codec=none",
|
||||
"session.timeout.ms=10000",
|
||||
"socket.timeout.ms=5000",
|
||||
"socket.keepalive.enable=true",
|
||||
"reconnect.backoff.jitter.ms=1000",
|
||||
"enable.partition.eof=false" ]
|
||||
)
|
||||
|
||||
template(name="outfmt" type="string" string="%msg:F,58:2%\n")
|
||||
|
||||
if ($msg contains "msgnum:") then {
|
||||
action( type="omfile" file=`echo $RSYSLOG_OUT_LOG` template="outfmt" )
|
||||
}
|
||||
'
|
||||
|
||||
# Start imkafka receiver config
|
||||
echo Starting receiver instance [imkafka]
|
||||
startup
|
||||
# ---
|
||||
|
||||
# Messure Starttime
|
||||
TIMESTART=$(date +%s.%N)
|
||||
|
||||
# --- Fill Kafka Server with messages
|
||||
# Can properly be done in a better way?!
|
||||
for i in {00000001..00100000}
|
||||
do
|
||||
echo " msgnum:$i" >> $RSYSLOG_OUT_LOG.in
|
||||
done
|
||||
|
||||
echo Inject messages into kafka
|
||||
cat $RSYSLOG_OUT_LOG.in | kafkacat -P -b localhost:29092 -t $RANDTOPIC
|
||||
# ---
|
||||
|
||||
echo Give imkafka some time to start...
|
||||
sleep 5
|
||||
|
||||
echo Stopping sender instance [omkafka]
|
||||
shutdown_when_empty
|
||||
wait_shutdown
|
||||
|
||||
# Messure Endtime
|
||||
TIMEEND=$(date +%s.%N)
|
||||
TIMEDIFF=$(echo "$TIMEEND - $TIMESTART" | bc)
|
||||
echo "*** imkafka time to process all data: $TIMEDIFF seconds!"
|
||||
|
||||
# Delete topic to remove old traces before
|
||||
. $srcdir/diag.sh delete-kafka-topic $RANDTOPIC '.dep_wrk' '22181'
|
||||
|
||||
echo \[sndrcv_kafka.sh\]: stop kafka instances
|
||||
. $srcdir/diag.sh stop-kafka
|
||||
. $srcdir/diag.sh stop-zookeeper
|
||||
|
||||
# Do the final sequence check
|
||||
seq_check 1 $TESTMESSAGESFULL -d
|
||||
|
||||
echo success
|
||||
exit_test
|
||||
Loading…
x
Reference in New Issue
Block a user