-
Notifications
You must be signed in to change notification settings - Fork 54
/
Copy pathproducer.py
executable file
·39 lines (32 loc) · 1.29 KB
/
producer.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
import sys
import os
from confluent_kafka import Producer
if __name__ == '__main__':
topic = os.environ['CLOUDKARAFKA_TOPIC'].split(",")[0]
# Consumer configuration
# See https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md
conf = {
'bootstrap.servers': os.environ['CLOUDKARAFKA_BROKERS'],
'session.timeout.ms': 6000,
'default.topic.config': {'auto.offset.reset': 'smallest'},
'security.protocol': 'SASL_SSL',
'sasl.mechanisms': 'SCRAM-SHA-256',
'sasl.username': os.environ['CLOUDKARAFKA_USERNAME'],
'sasl.password': os.environ['CLOUDKARAFKA_PASSWORD']
}
p = Producer(**conf)
def delivery_callback(err, msg):
if err:
sys.stderr.write('%% Message failed delivery: %s\n' % err)
else:
sys.stderr.write('%% Message delivered to %s [%d]\n' %
(msg.topic(), msg.partition()))
for line in sys.stdin:
try:
p.produce(topic, line.rstrip(), callback=delivery_callback)
except BufferError as e:
sys.stderr.write('%% Local producer queue is full (%d messages awaiting delivery): try again\n' %
len(p))
p.poll(0)
sys.stderr.write('%% Waiting for %d deliveries\n' % len(p))
p.flush()