修改代码

src/main/java/com/dsk/flink/dsc/sync/SyncCustomerDataSource.java

@@ -45,13 +45,17 @@ public class SyncCustomerDataSource {
 
         //获取用户自己的配置信息
         ParameterTool parameterTool = ParameterTool.fromArgs(args);
+        String offsetTimestamp = parameterTool.get("offsetTimestamp");
         String propertiesPath = parameterTool.get("propertiesPath");
         EnvProperties envProps = EnvPropertiesUtil.getPropertiesFromArgsPath(propertiesPath);
         envProps.put("providerImpl", JdbcConnectionProviderFactory.HikariDataSourceJdbcConnectionProvider.class.getName());
         //System.out.println("读取到的配置文件：-> " + envProps.toString());
 
         FlinkKafkaConsumer<String> kafkaConsumer = new FlinkKafkaConsumer<String>(envProps.getKafka_topic(), new SimpleStringSchema(), EtlUtils.getKafkaConfig(envProps.getKafka_brokers(), envProps.getKafka_topic().concat("-group")));
-        //kafkaConsumer.setStartFromEarliest();
+        //lww添加
+        if (StrUtil.isNotBlank(offsetTimestamp)) {
+            kafkaConsumer.setStartFromTimestamp(Long.parseLong(offsetTimestamp));
+        }
 
         SingleOutputStreamOperator<String> kafkaSource = env.addSource(kafkaConsumer)
                 .setParallelism(1)

src/main/java/com/dsk/flink/dsc/utils/EnvPropertiesUtil.java

@@ -21,7 +21,7 @@ import java.util.zip.ZipEntry;
  */
 public class EnvPropertiesUtil {
 
-	private static String SYSTEM_ENV = "dev";
+	private static String SYSTEM_ENV = "pro";
 	private static EnvProperties properties = new EnvProperties();
 	static {
 		SYSTEM_ENV = System.getProperty("profiles.active") == null ? SYSTEM_ENV : System.getProperty("profiles.active");

src/main/java/com/dsk/flink/dsc/utils/EtlUtils.java

@@ -10,8 +10,8 @@ public class EtlUtils {
         properties.setProperty("group.id", groupId);
         properties.setProperty("key.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
         properties.setProperty("value.deserializer", "org.apache.kafka.common.serialization.StringDeserializer");
-        //properties.setProperty("auto.offset.reset", "earliest");
-        properties.setProperty("auto.offset.reset", "latest");
+        properties.setProperty("auto.offset.reset", "earliest");
+        //properties.setProperty("auto.offset.reset", "latest");
         //properties.setProperty("max.poll.interval.ms", "604800000");
         //properties.setProperty("session.timeout.ms", "20160000");
         //properties.setProperty("heartbeat.interval.ms", "6720000");
@@ -20,7 +20,7 @@ public class EtlUtils {
         /*properties.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.plain.PlainLoginModule required username=\"dsk\" password=\"LU1SRhTzoxssRoCp\";");
         properties.setProperty("security.protocol", "SASL_PLAINTEXT");
         properties.setProperty("sasl.mechanism", "PLAIN");*/
-        properties.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"dsk_st\" password=\"vIfVv6esjpwU2jT\";");
+        properties.setProperty("sasl.jaas.config", "org.apache.kafka.common.security.scram.ScramLoginModule required username=\"ccroad\" password=\"BkYeyyb9KkFGkVua9sjzwWZVL\";");
         properties.setProperty("security.protocol", "SASL_PLAINTEXT");
         properties.setProperty("sasl.mechanism", "SCRAM-SHA-512");
         return properties;