Kafka, Kafka Consumer, Kafka Producer, Confluent Kafka, Streaming.
skip the header for row in reader: records.append(', '.join(row)) return records def delivery_report(err, msg): if err is not None: print("Delivery failed for record {}: {}".format(msg.key(), err)) return print('Record:{} successfully produced to topic:{} partition:[{}] at offset:{}'.format(