From 4587c97511b7f5ec740c4abddab5f9698b8fc4bb Mon Sep 17 00:00:00 2001
From: "sky.huang" <sky.huang@1data.info>
Date: Wed, 14 Jun 2023 11:46:33 +0800
Subject: [PATCH] =?UTF-8?q?=E5=8D=87=E7=BA=A7=20Flink=20=E7=89=88=E6=9C=AC?=
 =?UTF-8?q?=E4=B8=BA=201.17.1=20&=20=E8=A7=A3=E5=86=B3=20Kafka=20=E4=BE=9D?=
 =?UTF-8?q?=E8=B5=96=E5=86=B2=E7=AA=81=E9=97=AE=E9=A2=98=20&=20Redis=20?=
 =?UTF-8?q?=E6=94=AF=E6=8C=81=20=E5=93=A8=E5=85=B5=E6=A8=A1=E5=BC=8F?=
 =?UTF-8?q?=E8=BF=9E=E6=8E=A5?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 austin-data-house/pom.xml                     |  4 ++--
 austin-stream/pom.xml                         | 11 +++++-----
 .../stream/constants/AustinFlinkConstant.java | 13 +++++++++++-
 .../stream/utils/LettuceRedisUtils.java       | 17 ++++++++++++----
 pom.xml                                       | 20 ++++++++++++-------
 5 files changed, 46 insertions(+), 19 deletions(-)

diff --git a/austin-data-house/pom.xml b/austin-data-house/pom.xml
index 30b5381..dba1fab 100644
--- a/austin-data-house/pom.xml
+++ b/austin-data-house/pom.xml
@@ -23,13 +23,13 @@
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-connector-hive_2.12</artifactId>
-            <version>1.16.0</version>
+            <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
             <artifactId>flink-table-api-java-bridge</artifactId>
-            <version>1.16.0</version>
+            <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
 
diff --git a/austin-stream/pom.xml b/austin-stream/pom.xml
index 3bc7137..f995bf6 100644
--- a/austin-stream/pom.xml
+++ b/austin-stream/pom.xml
@@ -15,26 +15,27 @@
     <dependencies>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
+            <artifactId>flink-walkthrough-common</artifactId>
             <version>${flink.version}</version>
         </dependency>
         <!-- This dependency is provided, because it should not be packaged into the JAR file. -->
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+            <artifactId>flink-streaming-java</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-clients_${scala.binary.version}</artifactId>
+            <artifactId>flink-clients</artifactId>
             <version>${flink.version}</version>
             <scope>provided</scope>
         </dependency>
         <!-- Add connector dependencies here. They must be in the default scope (compile). -->
+
         <dependency>
             <groupId>org.apache.flink</groupId>
-            <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+            <artifactId>flink-connector-kafka</artifactId>
             <version>${flink.version}</version>
         </dependency>
 
@@ -85,7 +86,7 @@
                                     <resource>META-INF/spring.schemas</resource>
                                 </transformer>
                                 <transformer
-                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer" />
+                                        implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                 <transformer
                                         implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                     <mainClass>com.java3y.austin.stream.AustinBootStrap</mainClass>
diff --git a/austin-stream/src/main/java/com/java3y/austin/stream/constants/AustinFlinkConstant.java b/austin-stream/src/main/java/com/java3y/austin/stream/constants/AustinFlinkConstant.java
index 7bf2262..d02afae 100644
--- a/austin-stream/src/main/java/com/java3y/austin/stream/constants/AustinFlinkConstant.java
+++ b/austin-stream/src/main/java/com/java3y/austin/stream/constants/AustinFlinkConstant.java
@@ -22,10 +22,21 @@ public class AustinFlinkConstant {
      * TODO 使用前配置redis ip:port
      * (真实网络ip,这里不能用配置的hosts，看语雀文档得到真实ip)
      */
+
+    public static final String REDIS_MODE_SENTINEL = "SENTINEL";
+
+    public static final String REDIS_MODE_SINGLE = "SINGLE";
+
+    public static final String REDIS_MODE = REDIS_MODE_SENTINEL;
+
     public static final String REDIS_IP = "austin-redis";
-    public static final String REDIS_PORT = "6379";
+
+    public static final Integer REDIS_PORT = 6379;
+
     public static final String REDIS_PASSWORD = "austin";
 
+    public static final String MASTER_ID = "mymaster";
+
 
     /**
      * Flink流程常量
diff --git a/austin-stream/src/main/java/com/java3y/austin/stream/utils/LettuceRedisUtils.java b/austin-stream/src/main/java/com/java3y/austin/stream/utils/LettuceRedisUtils.java
index 0c64cd2..7c9269a 100644
--- a/austin-stream/src/main/java/com/java3y/austin/stream/utils/LettuceRedisUtils.java
+++ b/austin-stream/src/main/java/com/java3y/austin/stream/utils/LettuceRedisUtils.java
@@ -1,5 +1,6 @@
 package com.java3y.austin.stream.utils;
 
+import cn.hutool.core.util.StrUtil;
 import com.java3y.austin.stream.callback.RedisPipelineCallBack;
 import com.java3y.austin.stream.constants.AustinFlinkConstant;
 import io.lettuce.core.LettuceFutures;
@@ -26,10 +27,18 @@ public class LettuceRedisUtils {
     private static RedisClient redisClient;
 
     static {
-        RedisURI redisUri = RedisURI.Builder.redis(AustinFlinkConstant.REDIS_IP)
-                .withPort(Integer.valueOf(AustinFlinkConstant.REDIS_PORT))
-                .withPassword(AustinFlinkConstant.REDIS_PASSWORD.toCharArray())
-                .build();
+        RedisURI redisUri = null;
+        if (StrUtil.equals(AustinFlinkConstant.REDIS_MODE_SENTINEL, AustinFlinkConstant.REDIS_MODE)) {
+            redisUri = RedisURI.Builder.sentinel(AustinFlinkConstant.REDIS_IP, AustinFlinkConstant.MASTER_ID)
+                    .withPassword(AustinFlinkConstant.REDIS_PASSWORD.toCharArray())
+                    .build();
+        } else if (StrUtil.equals(AustinFlinkConstant.REDIS_MODE_SINGLE, AustinFlinkConstant.REDIS_MODE)) {
+            redisUri = RedisURI.Builder.redis(AustinFlinkConstant.REDIS_IP)
+                    .withPort(Integer.valueOf(AustinFlinkConstant.REDIS_PORT))
+                    .withPassword(AustinFlinkConstant.REDIS_PASSWORD.toCharArray())
+                    .build();
+        }
+
         redisClient = RedisClient.create(redisUri);
     }
 
diff --git a/pom.xml b/pom.xml
index 7692650..80e2a69 100644
--- a/pom.xml
+++ b/pom.xml
@@ -32,9 +32,8 @@
     <properties>
         <java.version>1.8</java.version>
         <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
-        <flink.version>1.14.3</flink.version>
+        <flink.version>1.17.1</flink.version>
         <target.java.version>1.8</target.java.version>
-        <scala.binary.version>2.11</scala.binary.version>
         <maven.compiler.source>${target.java.version}</maven.compiler.source>
         <maven.compiler.target>${target.java.version}</maven.compiler.target>
         <log4j.version>2.17.1</log4j.version>
@@ -54,7 +53,7 @@
             <dependency>
                 <groupId>cn.hutool</groupId>
                 <artifactId>hutool-all</artifactId>
-                <version>5.7.15</version>
+                <version>5.8.16</version>
             </dependency>
 
             <!--guava工具包-->
@@ -135,21 +134,28 @@
                 <version>2.3.0</version>
             </dependency>
 
+            <!--Kafka 消息队列-->
+            <dependency>
+                <groupId>org.apache.kafka</groupId>
+                <artifactId>kafka-clients</artifactId>
+                <version>3.2.3</version>
+            </dependency>
+
             <!--flink相关依赖-->
             <dependency>
                 <groupId>org.apache.flink</groupId>
-                <artifactId>flink-walkthrough-common_${scala.binary.version}</artifactId>
+                <artifactId>flink-walkthrough-common</artifactId>
                 <version>${flink.version}</version>
             </dependency>
             <dependency>
                 <groupId>org.apache.flink</groupId>
-                <artifactId>flink-streaming-java_${scala.binary.version}</artifactId>
+                <artifactId>flink-streaming-java</artifactId>
                 <version>${flink.version}</version>
                 <scope>provided</scope>
             </dependency>
             <dependency>
                 <groupId>org.apache.flink</groupId>
-                <artifactId>flink-clients_${scala.binary.version}</artifactId>
+                <artifactId>flink-clients</artifactId>
                 <version>${flink.version}</version>
                 <scope>provided</scope>
             </dependency>
@@ -157,7 +163,7 @@
             <!--kafkaFlink连接器-->
             <dependency>
                 <groupId>org.apache.flink</groupId>
-                <artifactId>flink-connector-kafka_${scala.binary.version}</artifactId>
+                <artifactId>flink-connector-kafka</artifactId>
                 <version>${flink.version}</version>
             </dependency>