# Flink Kafka样例程序开发思路
#### 场景说明
假定某个Flink业务每秒就会收到1个消息记录。
基于某些业务要求，开发的Flink应用程序实现功能：实时输出带有前缀的消息内容。
 #### 数据规划
Flink样例工程的数据存储在Kafka组件中。向Kafka组件发送数据（需要有Kafka权限用户），并从Kafka组件接收数据。
1. 确保集群安装完成，包括HDFS、Yarn、Flink和Kafka。
2. 创建Topic。
   1. 在服务端配置用户创建topic的权限。 将Kafka的Broker配置参数"allow.everyone.if.no.acl.found"的值修改为"true"，如[图1]所示。配置完后重启Kafka服务。
      图1配置用户创建topic的权限   
      ![](https://support.huaweicloud.com/devg3-mrs/zh-cn_image_0263897478.png) 
   
   2. 用户使用Linux命令行创建topic，执行命令前需要使用kinit命令进行人机认证，如：**kinit** *flinkuse* r。
      ![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/note_3.0-zh-cn.png)
      flinkuser需要用户自己创建，并拥有创建Kafka的topic权限。具体操作请参考[准备MRS应用开发用户](https://support.huaweicloud.com/devg3-mrs/mrs_07_020002.html)章节。
      创建topic的命令格式：
      **bin/kafka-topics.sh --create --zookeeper {** *zkQuorum* **}/kafka --partitions {** *partitionNum* **} --replication-factor {** *replicationNum* **} --topic {** *Topic* **}**
      表1参数说明 
      | **参数名**          | **说明**                    |
      |:---|:---|
      | {zkQuorum}       | ZooKeeper集群信息，格式为IP:port。 |
      | {partitionNum}   | topic的分区数。                |
      | {replicationNum} | topic中每个partition数据的副本数。  |
      | {Topic}          | Topic名称。                  |
         
      示例：在Kafka的客户端路径下执行命令，此处以ZooKeeper集群的IP:port是10.96.101.32:2181,10.96.101.251:2181,10.96.101.177:2181,10.91.8.160:2181，Topic名称为topic1的数据为例。
      ```
      bin/kafka-topics.sh --create --zookeeper 10.96.101.32:2181,10.96.101.251:2181,10.96.101.177:2181,10.91.8.160:2181/kafka --partitions 5 --replication-factor 1 --topic topic1
      ```
      
    
3. 安全认证。 安全认证的方式有三种：Kerberos认证、SSL加密认证和Kerberos+SSL模式认证，用户在使用的时候可任选其中一种方式进行认证。
   - **Kerberos认证配置**
     1. 客户端配置。 在Flink配置文件"flink-conf.yaml"中，增加kerberos认证相关配置（主要在"contexts"项中增加"KafkaClient"），示例如下：
        ```
        security.kerberos.login.keytab: /home/demo/flink/release/flink-x.x.x/keytab/user.keytab
        security.kerberos.login.principal: flinkuser
        security.kerberos.login.contexts: Client,KafkaClient
        security.kerberos.login.use-ticket-cache: false
        ```
        
     
     2. 运行参数。 关于"SASL_PLAINTEXT"协议的运行参数示例如下：
        ```
        --topic topic1 --bootstrap.servers 10.96.101.32:21007 --security.protocol SASL_PLAINTEXT  --sasl.kerberos.service.name kafka --kerberos.domain.name hadoop.系统域名.com
        ```
        ![](https://support.huaweicloud.com/devg3-mrs/public_sys-resources/note_3.0-zh-cn.png)
        - 10.96.101.32:21007：kafka服务器的IP和端口。
        
        - 系统域名：用户可登录FusionInsight Manager，单击"系统 \> 权限 \> 域和互信"，查看"本端域"参数，即为当前系统域名。
          
      
   
   - **SSL加密配置**
     - 服务端配置。 配置"ssl.mode.enable"为"true"，如[图2]所示：
       图2服务端配置   
       ![](https://support.huaweicloud.com/devg3-mrs/zh-cn_image_0263897542.png) 
     
     - 客户端配置。
       1. 登录FusionInsight Manager系统，选择"集群 \> *待操作集群的名称* \> 服务 \> Kafka \> 更多 \> 下载客户端" ，下载客户端压缩文件到本地机器。如[图3]所示：
          图3客户端配置   
          ![](https://support.huaweicloud.com/devg3-mrs/zh-cn_image_0263897492.png) 
       
       
       
       2. 使用客户端根目录中的"ca.crt"证书文件生成客户端的"truststore"。
          执行命令如下：
          ```
          keytool -noprompt -import -alias myservercert -file ca.crt -keystore truststore.jks
          ```
          命令执行结果查看：
          图4执行结果   
          ![](https://support.huaweicloud.com/devg3-mrs/zh-cn_image_0263897438.png) 
       
       3. 运行参数。 "ssl.truststore.password"参数内容需要跟创建"truststore"时输入的密码保持一致，执行以下命令运行参数。
          ```
          --topic topic1 --bootstrap.servers 10.96.101.32:9093 --security.protocol SSL --ssl.truststore.location /home/zgd/software/FusionInsight_XXX_Kafka_ClientConfig/truststore.jks --ssl.truststore.password xxx //10.96.101.32:9093表示kafka服务器的IP:port，XXX表示FusionInsight相应的版本号，xxx表示密码。
          ```
          
        
      
   
   - **Kerberos+SSL模式配置**
     完成上文中Kerberos和SSL各自的服务端和客户端配置后，只需要修改运行参数中的端口号和协议类型即可启动Kerberos+SSL模式。
     ```
     --topic topic1 --bootstrap.servers 10.96.101.32:21009 --security.protocol SASL_SSL  --sasl.kerberos.service.name kafka --ssl.truststore.location --kerberos.domain.name hadoop.系统域名.com /home/zgd/software/FusionInsight_XXX_Kafka_ClientConfig/truststore.jks --ssl.truststore.password xxx //10.96.101.32:21009表示kafka服务器的IP:port，XXX表示FusionInsight相应的版本号，xxx表示密码。
     ```
     
    
 
#### 开发思路
1. 启动Flink Kafka Producer应用向Kafka发送数据。
2. 启动Flink Kafka Consumer应用从Kafka接收数据，保证topic与producer一致。
3. 在数据内容中增加前缀并进行打印。