1849 support ssl to mce cli.py

metadata-ingestion/mce-cli/kafka.config

@@ -0,0 +1,14 @@
+# https://github.com/edenhill/librdkafka/wiki/Using-SSL-with-librdkafka
+# Kafka Brokers Certs
+#security.protocol=ssl
+ssl.ca.location=certificate_ca.pem
+ssl.certificate.location=certificate.pem
+ssl.key.location=private_key.pem
+ssl.key.password=XXXXXXX
+
+# Schema Registry Certs
+schema.registry.ssl.ca.location=certificate_ca.pem
+schema.registry.ssl.certificate.location=certificate.pem
+schema.registry.ssl.key.location=private_key.pem
+schema.registry.ssl.key.password=XXXXXXX
+

metadata-ingestion/mce-cli/mce_cli.py

@@ -78,11 +78,30 @@ def consume(conf, schema_record):
     c.close()
 
 
+def convert_config_to_dict(file_location):
+    #print("converting: " + file_location)
+    result = {}
+    f = open(file_location, "r")
+    fl =f.readlines()
+    for line in fl:
+        if line[0] == '#' or line == '\n':
+            continue
+        key, value = line.split("=", 1)
+        if value is not None:
+            result = {**result, **{key : value.rsplit("\n")[0]}}
+        else:
+            result = {**result, **{key : None}}
+    return result
+
+
 def main(args):
     # Handle common configs
     conf = {'bootstrap.servers': args.bootstrap_servers,
             'schema.registry.url': args.schema_registry}
 
+    if args.command_config is not None:
+        conf.update(convert_config_to_dict(args.command_config))
+
     if args.mode == "produce":
         produce(conf, args.data_file, args.schema_record)
     else:
@@ -104,4 +123,7 @@ def main(args):
                         help="Avro schema record; required if running 'producer' mode")
     parser.add_argument('-d', dest="data_file", default="bootstrap_mce.dat",
                         help="MCE data file; required if running 'producer' mode")
+    parser.add_argument('-c', dest="command_config",
+                        default="None", help="Kafka SSL details")
     main(parser.parse_args())
+