package com.baidu.inf.flink
 
import java.util.Properties
 
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.streaming.api.scala.{StreamExecutionEnvironment, _}
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.kafka.common.serialization.StringDeserializer
import org.slf4j.LoggerFactory
 
object CloudFlinkConsumeKafkaDemo {
  private val logger = LoggerFactory.getLogger(this.getClass)
 
  def main(args: Array[String]): Unit = {
    logger.info("************ Flink Consume Kafka Demo start **************")
    if (args.length < 7) {
      logger.error(" Parameters Are Missing , " +
        "Needs : <topic> " +
        "<groupId> " +
        "<brokerHosts> " +
        "<truststore_location> " +
        "<truststore_pass> " +
        "<keystore_location> " +
        "<keystore_pass>")
      System.exit(-1)
    }
    val Array(topic, groupId, brokerHosts,
    truststore_location, truststore_pass,
    keystore_location, keystore_pass, _*) = args
 
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(2)
    env.getConfig.disableSysoutLogging
 
    val kafkaProperties = new Properties()
    kafkaProperties.setProperty("bootstrap.servers", brokerHosts)
    kafkaProperties.setProperty("key.deserializer", classOf[StringDeserializer].getName)
    kafkaProperties.setProperty("value.deserializer", classOf[StringDeserializer].getName)
    kafkaProperties.setProperty("group.id", groupId)
    kafkaProperties.setProperty("auto.offset.reset", "latest")
    kafkaProperties.setProperty("serializer.class", "kafka.serializer.StringEncoder")
    kafkaProperties.setProperty("security.protocol", "SSL")
    kafkaProperties.setProperty("ssl.truststore.location", truststore_location)
    kafkaProperties.setProperty("ssl.truststore.password", truststore_pass)
    kafkaProperties.setProperty("ssl.keystore.location", keystore_location)
    kafkaProperties.setProperty("ssl.keystore.password", keystore_pass)
    kafkaProperties.setProperty("enable.auto.commit", "true")
 
    val ds = env.addSource(
      new FlinkKafkaConsumer[String](topic,
        new SimpleStringSchema(),
        kafkaProperties))
 
    ds.print()
    env.execute()
  }
}