dnwe commented on a change in pull request #202: Produce action fixes URL: https://github.com/apache/incubator-openwhisk-package-kafka/pull/202#discussion_r131248585
########## File path: action/kafkaProduce.py ########## @@ -18,46 +18,96 @@ */ """ -import ssl import base64 +import logging +import sys +import traceback + from kafka import KafkaProducer -from kafka.errors import NoBrokersAvailable +from kafka.errors import NoBrokersAvailable, KafkaTimeoutError, AuthenticationFailedError +from kafka.version import __version__ +from random import shuffle + +logging.basicConfig(stream=sys.stdout, level=logging.INFO, + format='%(levelname)-8s %(asctime)s %(message)s', + datefmt='[%H:%M:%S]') + +max_cached_producers = 10 def main(params): + producer = None + logging.info("Using kafka-python %s", str(__version__)) + + print("Validating parameters") validationResult = validateParams(params) if validationResult[0] != True: return {'error': validationResult[1]} else: validatedParams = validationResult[1] - brokers = params['brokers'] - - try: - producer = KafkaProducer( - api_version_auto_timeout_ms=15000, - bootstrap_servers=brokers) - - print("Created producer") + attempt = 0 + max_attempts = 3 - # only use the key parameter if it is present - if 'key' in validatedParams: - messageKey = validatedParams['key'] - producer.send(validatedParams['topic'], bytes(validatedParams['value'], 'utf-8'), key=bytes(messageKey, 'utf-8')) - else: - producer.send(validatedParams['topic'], bytes(validatedParams['value'], 'utf-8')) + result = {"success": True} - producer.flush() + while attempt < max_attempts: + attempt += 1 + print("Starting attempt {}".format(attempt)) - print("Sent message") - except NoBrokersAvailable: - # this exception's message is a little too generic + try: + print("Getting producer") + producer = getProducer(validatedParams) + + topic = validatedParams['topic'] + print("Finding topic {}".format(topic)) + partition_info = producer.partitions_for(topic) + print("Found topic {} with partition(s) {}".format(topic, partition_info)) + + break + except Exception as e: + if attempt == max_attempts: + producer = None + logging.warning(e) + traceback.print_stack(limit=5) + result = getResultForException(e) + + # we successfully connected and found the topic metadata... let's send! + if producer is not None: + try: + print("Producing message") + + # only use the key parameter if it is present + value = validatedParams['value'] + if 'key' in validatedParams: + messageKey = validatedParams['key'] + future = producer.send( + topic, bytes(value, 'utf-8'), key=bytes(messageKey, 'utf-8')) + else: + future = producer.send(topic, bytes(value, 'utf-8')) + + sent = future.get(timeout=20) + msg = "Successfully sent message to {}:{} at offset {}".format( + sent.topic, sent.partition, sent.offset) + print(msg) + result = {"success": True, "message": msg} + except Exception as e: + logging.warning(e) + traceback.print_stack(limit=5) Review comment: As above, you want `print_exc` here I think. ---------------------------------------------------------------- This is an automated message from the Apache Git Service. To respond to the message, please log on GitHub and use the URL above to go to the specific comment. For queries about this service, please contact Infrastructure at: us...@infra.apache.org With regards, Apache Git Services