Warn when INCLUDE_SPARK_PROTOBUF_JAR=true but jars are missing

Surface a stderr warning when the variable is explicitly requested but
the spark-protobuf/protobuf-java jars are not present, so a CI
misconfiguration is not masked as a silent skip. Default opt-out
(unset or false) stays silent.

Addresses greptile review feedback on #14885.

Signed-off-by: Haoyang Li <haoyangl@nvidia.com>

Author: thirtiseven

integration_tests/run_pyspark_from_build.sh

@@ -150,11 +150,15 @@ else
     # the unshaded `protobuf-java` come from maven-dependency-plugin and must both be present
     # -- spark-protobuf shades `com.google.protobuf.*` internally and Spark does not bundle
     # the unshaded jar.
-    if [[ $( echo ${INCLUDE_SPARK_PROTOBUF_JAR} | tr '[:upper:]' '[:lower:]' ) != "false" \
+    INCLUDE_SPARK_PROTOBUF_JAR_REQUESTED=$(echo "${INCLUDE_SPARK_PROTOBUF_JAR}" | tr '[:upper:]' '[:lower:]')
+    if [[ "$INCLUDE_SPARK_PROTOBUF_JAR_REQUESTED" != "false" \
           && $(readlink -e $PROTOBUF_JARS 2>/dev/null | wc -l) -eq 2 ]];
     then
         export INCLUDE_SPARK_PROTOBUF_JAR=true
     else
+        if [[ "$INCLUDE_SPARK_PROTOBUF_JAR_REQUESTED" == "true" ]]; then
+            >&2 echo "WARNING: INCLUDE_SPARK_PROTOBUF_JAR=true was requested but spark-protobuf/protobuf-java jars were not found under $TARGET_DIR/dependency; disabling protobuf tests."
+        fi
         export INCLUDE_SPARK_PROTOBUF_JAR=false
         PROTOBUF_JARS=""
     fi

integration_tests/src/main/python/protobuf_test.py

@@ -14,6 +14,7 @@
 
 import inspect
 import os
+import tempfile
 
 import pytest
 
@@ -88,23 +89,17 @@ def _build_simple_descriptor_bytes(spark):
     return bytes(fds.toByteArray())
 
 
-def _write_bytes_to_hadoop_path(spark, path_str, data_bytes):
-    sc = spark.sparkContext
-    config = sc._jsc.hadoopConfiguration()
-    jpath = sc._jvm.org.apache.hadoop.fs.Path(path_str)
-    fs = sc._jvm.org.apache.hadoop.fs.FileSystem.get(config)
-    out = fs.create(jpath, True)
-    try:
-        out.write(bytearray(data_bytes))
-    finally:
-        out.close()
-
-
-def _setup_simple_desc(spark_tmp_path):
-    desc_path = spark_tmp_path + "/simple.desc"
+@pytest.fixture
+def simple_desc(request):
+    # spark-protobuf reads descFilePath with `new File(...)` + FileUtils
+    # (driver-local), not via Hadoop FileSystem -- so the descriptor must live
+    # on the driver's local disk. Writing to `spark_tmp_path` would break in
+    # distributed setups where it resolves to HDFS / GCS / etc.
     desc_bytes = with_cpu_session(_build_simple_descriptor_bytes)
-    with_cpu_session(
-        lambda spark: _write_bytes_to_hadoop_path(spark, desc_path, desc_bytes))
+    fd, desc_path = tempfile.mkstemp(suffix=".desc")
+    with os.fdopen(fd, "wb") as fp:
+        fp.write(desc_bytes)
+    request.addfinalizer(lambda: os.path.exists(desc_path) and os.unlink(desc_path))
     return desc_path, desc_bytes
 
 
@@ -117,8 +112,8 @@ def _make_smoke_df(spark):
 
 
 @allow_non_gpu("ProjectExec", "ProtobufDataToCatalyst")
-def test_from_protobuf_smoke_path_api(spark_tmp_path, from_protobuf_fn):
-    desc_path, _ = _setup_simple_desc(spark_tmp_path)
+def test_from_protobuf_smoke_path_api(simple_desc, from_protobuf_fn):
+    desc_path, _ = simple_desc
 
     def run(spark):
         return _make_smoke_df(spark).select(
@@ -128,10 +123,10 @@ def run(spark):
 
 
 @allow_non_gpu("ProjectExec", "ProtobufDataToCatalyst")
-def test_from_protobuf_smoke_binary_descriptor_api(spark_tmp_path, from_protobuf_fn):
+def test_from_protobuf_smoke_binary_descriptor_api(simple_desc, from_protobuf_fn):
     if "binaryDescriptorSet" not in inspect.signature(from_protobuf_fn).parameters:
         pytest.skip("binaryDescriptorSet kwarg is Spark 3.5+ only")
-    _, desc_bytes = _setup_simple_desc(spark_tmp_path)
+    _, desc_bytes = simple_desc
 
     def run(spark):
         return _make_smoke_df(spark).select(