[hotfix] Not throw exception when fail to list splits for streaming read partitioned table (apache#996)

Author: luoyuxia

fluss-flink/fluss-flink-common/src/main/java/com/alibaba/fluss/flink/source/enumerator/FlinkSourceEnumerator.java

@@ -524,8 +524,16 @@ private void handlePartitionsRemoved(Collection<Partition> removedPartitionInfo)
 
     private void handleSplitsAdd(List<SourceSplitBase> splits, Throwable t) {
         if (t != null) {
-            throw new FlinkRuntimeException(
-                    String.format("Failed to list splits for %s to read due to ", tablePath), t);
+            if (isPartitioned && streaming && scanPartitionDiscoveryIntervalMs > 0) {
+                // it means continuously read new partition splits, not throw exception, temporally
+                // warn it to avoid job fail. TODO: fix me in #288
+                LOG.warn("Failed to list splits for {}.", tablePath, t);
+                return;
+            } else {
+                throw new FlinkRuntimeException(
+                        String.format("Failed to list splits for %s to read due to ", tablePath),
+                        t);
+            }
         }
         if (isPartitioned) {
             if (!streaming || scanPartitionDiscoveryIntervalMs <= 0) {