接入kafka数据流需要增加groupID参数

realtime/hologram-streaming/src/main/java/com/zhht/irn/utils/FlinkUtils.java

@@ -48,8 +48,13 @@ public class FlinkUtils {
 
     public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String offset) {
         ParameterTool kafkaTool = initParameterToolByArgs(args);
+        String groupId =  kafkaTool.get("group.id", "test");
+        return createKafkaStream(args, environment, topic, groupId, offset);
+    }
+
+    public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String groupId, String offset) {
+        ParameterTool kafkaTool = initParameterToolByArgs(args);
         // 加载Kafka信息 { topic 相关 }
-        String groupId = kafkaTool.get("group.id", "group-wuyanfei");
         String servers = kafkaTool.getRequired("bootstrap.servers");
         List<String> topics = Arrays.asList(topic.split(","));
         String autoCommit = kafkaTool.get("enable.auto.commit", "false");
@@ -75,6 +80,8 @@ public class FlinkUtils {
         return rawStringStream;
     }
 
+
+
     private static ParameterTool initParameterToolByArgs(String[] args) {
         // 获取Kafka配置文件
         ParameterTool finkArgs = ParameterTool.fromArgs(args);

realtime/hologram-streaming/src/main/resources/kafka.properties

 bootstrap.servers=10.100.32.173:9092
 kafka.input.topics=trips_info
 auto.offset.reset=earliest
+group.id=test
\ No newline at end of file