Update test and docs to mention that only Spark 3 is affected

Author: razvan

docs/modules/spark-k8s/pages/usage-guide/spark-connect.adoc

@@ -198,9 +198,9 @@ The following features are not supported by the Stackable Spark operator yet
 
 == Known Issues
 
-* Distributed operations on Apache Iceberg tables fail on the executors.
+* Distributed operations on Apache Iceberg tables fail on the executors *with Spark 3.5.x*.
   PySpark calls that trigger a distributed job over an Iceberg table -- for example `DataFrame.collect()` or `DataFrame.count()` -- fail with `java.lang.ClassCastException: cannot assign instance of java.lang.invoke.SerializedLambda to field org.apache.spark.rdd.MapPartitionsRDD.f of type scala.Function3`.
   Driver-only operations still work: DDL (`CREATE`/`DROP`), `INSERT` of small data, `DataFrame.show()` of a small result, `SHOW TABLES` and `DESCRIBE`.
   The cause is upstream in Spark Connect: executor tasks run under a per-session class loader (which fetches session classes from the driver's artifact server) that cannot deserialize the Iceberg scan closures.
   It is independent of how the Iceberg runtime is provisioned -- it reproduces with `--packages`, with the jar placed on the system class path (`/stackable/spark/jars`), with `spark.addArtifact()`, and with combinations of these.
-  See https://issues.apache.org/jira/browse/SPARK-46032[SPARK-46032] (open) and https://issues.apache.org/jira/browse/SPARK-51537[SPARK-51537] for details.
+  *This is fixed in Spark 4* (https://issues.apache.org/jira/browse/SPARK-51537[SPARK-51537] / https://github.com/apache/spark/pull/50475[apache/spark#50475]), where distributed reads and writes of Iceberg tables work; it is not backported to the 3.5.x line (see the still-open https://issues.apache.org/jira/browse/SPARK-46032[SPARK-46032]).

tests/templates/kuttl/spark-connect-kerberos/20-run-connect-client.yaml.j2

@@ -57,25 +57,30 @@ data:
       spark.sql("DROP TABLE IF EXISTS iceberg.lakehouse.greetings")
       spark.sql("CREATE TABLE iceberg.lakehouse.greetings (id INT, data STRING) USING iceberg")
       spark.sql("INSERT INTO iceberg.lakehouse.greetings VALUES (1, 'one'), (2, 'two'), (3, 'three')")
-      spark.sql("SELECT * FROM iceberg.lakehouse.greetings ORDER BY id").show()
 
-      # Verify the Iceberg table is registered in the kerberized metastore via a
-      # metastore query (driver-side). We deliberately avoid result.collect() /
-      # count() over the Iceberg table: a *distributed* scan ships the Iceberg
-      # RDD closure to the executors, where it currently fails to deserialize on
-      # Spark Connect ("cannot assign SerializedLambda ... Function3"). This is an
-      # upstream Spark Connect classloader limitation (see SPARK-46032 /
-      # SPARK-51537) independent of how the Iceberg jar is provisioned - it
-      # reproduces with --packages, with the jar on the system classpath, with
-      # spark.addArtifact(), and with --packages + addArtifact together - and is
-      # separate from the metastore/Kerberos integration this test validates.
-      ice_tables = [row.tableName for row in spark.sql("SHOW TABLES IN iceberg.lakehouse").collect()]
-      print("Tables in 'iceberg.lakehouse':", ice_tables)
-      assert "greetings" in ice_tables, f"Iceberg table not registered in the HMS: {ice_tables}"
-
-      ice_columns = [row.col_name for row in spark.sql("DESCRIBE TABLE iceberg.lakehouse.greetings").collect()]
-      print("Columns of iceberg.lakehouse.greetings:", ice_columns)
-      assert "id" in ice_columns and "data" in ice_columns, f"unexpected columns: {ice_columns}"
+      # Reading the Iceberg table back differs by Spark version:
+      #
+      # Spark 4 includes the Spark Connect executor class loader fix
+      # (SPARK-51537 / apache/spark#50475), so a *distributed* read works:
+      # collect() ships the Iceberg scan closure to the executors and succeeds.
+      #
+      # On Spark 3.5.x that distributed scan still fails on the executors with
+      # "cannot assign SerializedLambda ... Function3" (SPARK-46032, not fixed in
+      # 3.5.x), so we verify the table via metastore queries (driver-side) instead.
+      print(f"Connected to Spark {spark.version}")
+      if spark.version.startswith("4"):
+        result = spark.sql("SELECT * FROM iceberg.lakehouse.greetings ORDER BY id")
+        result.show()
+        rows = result.collect()
+        assert len(rows) == 3, f"expected 3 rows in the Iceberg table, got {len(rows)}"
+      else:
+        spark.sql("SELECT * FROM iceberg.lakehouse.greetings ORDER BY id").show()
+        ice_tables = [row.tableName for row in spark.sql("SHOW TABLES IN iceberg.lakehouse").collect()]
+        print("Tables in 'iceberg.lakehouse':", ice_tables)
+        assert "greetings" in ice_tables, f"Iceberg table not registered in the HMS: {ice_tables}"
+        ice_columns = [row.col_name for row in spark.sql("DESCRIBE TABLE iceberg.lakehouse.greetings").collect()]
+        print("Columns of iceberg.lakehouse.greetings:", ice_columns)
+        assert "id" in ice_columns and "data" in ice_columns, f"unexpected columns: {ice_columns}"
 
       print("[SUCCESS] Read/write against the kerberized Hive metastore and Iceberg catalog succeeded")