support kafka with security certification

[Nicole00]

close #185

support kerberos, SASL, SASL_PLAINTEXT.

How to use:

• if the kafka is not security certificated, then just config the following common configs:

    service:"127.0.0.1:9092"
    topic:"test"
    interval.seconds:10
    startingOffsets:earliest
    maxOffsetsPerTrigger:0
  

• if the kafka is security certificated by Kerberos, then add the following configs on the base of common configs:

  securityProtocol:SASL_PLAINTEXT
  mechanism:GASSAPI
  kerberos:true
  kerberosServiceName:kafka
  

When submit the exchange to spark, we need to config the kafka_client_jaas.conf and the krb5.conf for driver and executor :

spark-submit --master xxx \
 --conf "spark.driver.extraJavaOptions=-Djava.security.auth.login.config=/path/kafka_client_jaas.conf -Djava.security.krb5.conf=/path/krb5.conf" \
--conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=path//kafka_client_jaas.conf -Djava.security.krb5.conf=path/krb5.conf" \
--files /local/path/to/kafka_client_jaas.conf,/local/path/to/kafka.keytab,/local/path/to/krb5.conf \
--class  com.vesoft.nebula.exchange.Exchange  \
nebula-exchange-2.4-3.0-SNAPSHOT.jar -c application.conf

• if the kafka is security certificated by other security protocol, then just add the following configs on the base of common configs:

  securityProtocol:SASL_PLAINTEXT
  mechanism:GASSAPI
  

When submit the exchange to spark, we need to config the kafka_client_jaas.conf and the krb5.conf for driver and executor :

 spark-submit --master xxx \
 --conf "spark.driver.extraJavaOptions=-Djava.security.auth.login.config=/path/kafka_client_jaas.conf" \
 --conf "spark.executor.extraJavaOptions=-Djava.security.auth.login.config=path//kafka_client_jaas.conf" \
 --files /local/path/to/kafka_client_jaas.conf \
 --class  com.vesoft.nebula.exchange.Exchange  \
 nebula-exchange-2.4-3.0-SNAPSHOT.jar -c application.conf

[Nicole00]

description for parameters:

securityProtocol:              optional, String type, no default value. the security protocol for kafka
mechanism:                     optional, String type, no default value. the sasl mechanism for kafka
kerberos:true                  optional, boolean type, default value is false. if the security mechanism is 'kerberos'
kerberosServiceName:           optional, String type, no default value. the kerberos service name,  if the kerberos config is true, then kerberosServiceName must be config too.