[hotfix-38914][kafka] Throw no-restart exception when there is no data in all partitions.

Signed-off-by: tiezhu <tiezhu@dtstack.com>

Author: FlechazoW

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/sample/OffsetFetcher.java

@@ -18,6 +18,7 @@
 
 package com.dtstack.flink.sql.source.kafka.sample;
 
+import com.dtstack.flink.sql.source.kafka.throwable.KafkaSamplingUnavailableException;
 import org.apache.flink.streaming.connectors.kafka.internals.KafkaTopicPartition;
 import org.apache.kafka.clients.consumer.ConsumerConfig;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
@@ -41,7 +42,7 @@ default OffsetMap seekOffset(Properties props, String topic) {
         try (KafkaConsumer<?, ?> consumer = new KafkaConsumer<>(props)) {
             OffsetMap offsetMap = fetchOffset(consumer, topic);
 
-            judgeKafkaSampleIsAvailable(offsetMap);
+            judgeKafkaSampleIsAvailable(offsetMap, topic);
 
             return offsetMap;
         }
@@ -53,7 +54,7 @@ default OffsetMap seekOffset(Properties props, String topic) {
      *
      * @param offsetMap offset map
      */
-    default void judgeKafkaSampleIsAvailable(OffsetMap offsetMap) {
+    default void judgeKafkaSampleIsAvailable(OffsetMap offsetMap, String topic) {
         boolean kafkaSampleIsAvailable = false;
         Map<KafkaTopicPartition, Long> latest = offsetMap.getLatest();
         Map<KafkaTopicPartition, Long> earliest = offsetMap.getEarliest();
@@ -68,8 +69,10 @@ default void judgeKafkaSampleIsAvailable(OffsetMap offsetMap) {
         }
 
         if (!kafkaSampleIsAvailable) {
-            throw new RuntimeException(
-                    "Kafka sample is unavailable because there is no data in all partitions");
+            throw new KafkaSamplingUnavailableException(
+                    String.format(
+                            "Kafka sampling of [%s] is unavailable because there is no data in all partitions",
+                            topic));
         }
     }
 

kafka-base/kafka-base-source/src/main/java/com/dtstack/flink/sql/source/kafka/throwable/KafkaSamplingUnavailableException.java

@@ -0,0 +1,35 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership.  The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package com.dtstack.flink.sql.source.kafka.throwable;
+
+import org.apache.flink.runtime.throwable.ThrowableAnnotation;
+import org.apache.flink.runtime.throwable.ThrowableType;
+import org.apache.flink.util.FlinkRuntimeException;
+
+/**
+ * @author tiezhu
+ * @since 2021/6/17 星期四
+ */
+@ThrowableAnnotation(ThrowableType.NonRecoverableError)
+public class KafkaSamplingUnavailableException extends FlinkRuntimeException {
+
+    public KafkaSamplingUnavailableException(String message) {
+        super(message);
+    }
+}