RepoMirrors/ceph
1
0
Fork 0
mirror of https://github.com/ceph/ceph synced 2025-03-20 09:16:59 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #51862 from yuvalif/wip-61520-reef

Browse Source 
reef: test/rgw/notifications: fix kafka consumer shutdown issue

Reviewed-by: Shilpa Jagannath <smanjara@redhat.com>
This commit is contained in:
Yuri Weinstein 2023-06-08 12:06:54 -04:00 committed by GitHub
commit 54dcabdd65
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 6 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/test/rgw/bucket_notification/test_bn.py
View File

@ -419,7 +419,10 @@ class KafkaReceiver(object):
port = 9093 port = 9093
while remaining_retries > 0: while remaining_retries > 0:
try: try:
self.consumer = KafkaConsumer(topic, bootstrap_servers = kafka_server+':'+str(port), security_protocol=security_type) self.consumer = KafkaConsumer(topic,
bootstrap_servers = kafka_server+':'+str(port),
security_protocol=security_type,
consumer_timeout_ms=16000)
print('Kafka consumer created on topic: '+topic) print('Kafka consumer created on topic: '+topic)
break break
except Exception as error: except Exception as error:
@ -448,7 +451,7 @@ def kafka_receiver_thread_runner(receiver):
while not receiver.stop: while not receiver.stop:
for msg in receiver.consumer: for msg in receiver.consumer:
receiver.events.append(json.loads(msg.value)) receiver.events.append(json.loads(msg.value))
timer.sleep(0.1) time.sleep(0.1)
log.info('Kafka receiver ended') log.info('Kafka receiver ended')
print('Kafka receiver ended') print('Kafka receiver ended')
except Exception as error: except Exception as error:
@ -468,6 +471,7 @@ def stop_kafka_receiver(receiver, task):
receiver.stop = True receiver.stop = True
task.join(1) task.join(1)
try: try:
receiver.consumer.unsubscribe()
receiver.consumer.close() receiver.consumer.close()
except Exception as error: except Exception as error:
log.info('failed to gracefuly stop Kafka receiver: %s', str(error)) log.info('failed to gracefuly stop Kafka receiver: %s', str(error))