[pinot] branch master updated: Fail spark application in case of no input files (#9105)

[jlli]

This is an automated email from the ASF dual-hosted git repository.

jlli pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pinot.git


The following commit(s) were added to refs/heads/master by this push:
     new 3c799cef75 Fail spark application in case of no input files (#9105)
3c799cef75 is described below

commit 3c799cef75c04bda6d24b96a5a86b63a8dc48360
Author: Kartik Khare <kharekar...@gmail.com>
AuthorDate: Wed Jul 27 03:44:24 2022 +0530

    Fail spark application in case of no input files (#9105)
    
    * Fail spark application in case of no input files
    
    * Update 
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-2.4/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark/SparkSegmentGenerationJobRunner.java
    
    Co-authored-by: Xiaotian (Jackie) Jiang 
<17555551+jackie-ji...@users.noreply.github.com>
    
    * Update 
pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-3.2/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark3/SparkSegmentGenerationJobRunner.java
    
    Co-authored-by: Xiaotian (Jackie) Jiang 
<17555551+jackie-ji...@users.noreply.github.com>
    
    Co-authored-by: Kartik Khare <kharekartik@Kartiks-MacBook-Pro.local>
    Co-authored-by: Xiaotian (Jackie) Jiang 
<17555551+jackie-ji...@users.noreply.github.com>
---
 .../ingestion/batch/spark/SparkSegmentGenerationJobRunner.java    | 8 ++++++++
 .../ingestion/batch/spark3/SparkSegmentGenerationJobRunner.java   | 7 +++++++
 2 files changed, 15 insertions(+)

diff --git 
a/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-2.4/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark/SparkSegmentGenerationJobRunner.java
 
b/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-2.4/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark/SparkSegmentGenerationJobRunner.java
index 88e3dd7671..ed73a2b719 100644
--- 
a/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-2.4/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark/SparkSegmentGenerationJobRunner.java
+++ 
b/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-2.4/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark/SparkSegmentGenerationJobRunner.java
@@ -191,6 +191,13 @@ public class SparkSegmentGenerationJobRunner implements 
IngestionJobRunner, Seri
       }
     }
 
+    if (filteredFiles.isEmpty()) {
+      throw new RuntimeException(
+          String.format("No file found in the input directory: %s matching 
includeFileNamePattern: %s,"
+                  + " excludeFileNamePattern: %s", _spec.getInputDirURI(), 
_spec.getIncludeFileNamePattern(),
+              _spec.getExcludeFileNamePattern()));
+    }
+
     LOGGER.info("Found {} files to create Pinot segments!", 
filteredFiles.size());
     try {
       JavaSparkContext sparkContext = 
JavaSparkContext.fromSparkContext(SparkContext.getOrCreate());
@@ -234,6 +241,7 @@ public class SparkSegmentGenerationJobRunner implements 
IngestionJobRunner, Seri
       if (jobParallelism <= 0 || jobParallelism > numDataFiles) {
         jobParallelism = numDataFiles;
       }
+
       JavaRDD<String> pathRDD = sparkContext.parallelize(pathAndIdxList, 
jobParallelism);
 
       final String pluginsInclude =
diff --git 
a/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-3.2/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark3/SparkSegmentGenerationJobRunner.java
 
b/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-3.2/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark3/SparkSegmentGenerationJobRunner.java
index 497e7dcc68..6300040d4e 100644
--- 
a/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-3.2/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark3/SparkSegmentGenerationJobRunner.java
+++ 
b/pinot-plugins/pinot-batch-ingestion/pinot-batch-ingestion-spark-3.2/src/main/java/org/apache/pinot/plugin/ingestion/batch/spark3/SparkSegmentGenerationJobRunner.java
@@ -191,6 +191,13 @@ public class SparkSegmentGenerationJobRunner implements 
IngestionJobRunner, Seri
       }
     }
 
+    if (filteredFiles.isEmpty()) {
+      throw new RuntimeException(
+          String.format("No file found in the input directory: %s matching 
includeFileNamePattern: %s,"
+                  + " excludeFileNamePattern: %s", _spec.getInputDirURI(), 
_spec.getIncludeFileNamePattern(),
+              _spec.getExcludeFileNamePattern()));
+    }
+
     LOGGER.info("Found {} files to create Pinot segments!", 
filteredFiles.size());
     try {
       JavaSparkContext sparkContext = 
JavaSparkContext.fromSparkContext(SparkContext.getOrCreate());


---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscr...@pinot.apache.org
For additional commands, e-mail: commits-h...@pinot.apache.org