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

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

@@ -42,14 +42,14 @@ public class FlinkUtils {
 
     public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic) {
         ParameterTool kafkaTool = initParameterToolByArgs(args);
-        String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
-        return createKafkaStream(args, environment, topic, offsetReset);
+        String groupId = kafkaTool.get("group.id", "test");
+        return createKafkaStream(args, environment, topic, groupId);
     }
 
-    public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String offset) {
+    public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String groupId) {
         ParameterTool kafkaTool = initParameterToolByArgs(args);
-        String groupId =  kafkaTool.get("group.id", "test");
-        return createKafkaStream(args, environment, topic, groupId, offset);
+        String offsetReset = kafkaTool.get("auto.offset.reset", "earliest");
+        return createKafkaStream(args, environment, topic, groupId, offsetReset);
     }
 
     public static DataStream<String> createKafkaStream(String[] args, StreamExecutionEnvironment environment, String topic, String groupId, String offset) {
@@ -67,7 +67,7 @@ public class FlinkUtils {
 
 
         int checkpointInterval = kafkaTool.getInt("checkpoint.interval", 5000);
-        String checkpointPath = kafkaTool.get("checkpoint.path", "file:///f/workspace/imooc-flink/state");
+        String checkpointPath = kafkaTool.get("checkpoint.path", "file:///f/workspace/flink/state");
 
         env.enableCheckpointing(checkpointInterval);
         env.setStateBackend(new FsStateBackend(checkpointPath));