(kudu) branch master updated: [spark] KuduSink: honor auth and encryption opts

[mg...@apache.org]

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

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


The following commit(s) were added to refs/heads/master by this push:
     new 102683f65 [spark] KuduSink: honor auth and encryption opts
102683f65 is described below

commit 102683f65af844d269565345d7d8c6ebc55c92c5
Author: Marton Greber <gr...@gmail.com>
AuthorDate: Fri Sep 4 12:11:19 2026 +0200

    [spark] KuduSink: honor auth and encryption opts
    
    KuduSink captured requireAuthentication and encryptionPolicy in its
    constructor but never passed them to the KuduContext it created, so the
    context always fell back to the defaults (no authentication required,
    OPTIONAL encryption). As a result streaming writes silently ignored the
    kudu.requireAuthentication and kudu.encryptionPolicy options, unlike the
    batch read path in KuduRelation which forwards them correctly.
    
    Forward both options into the KuduContext, matching KuduRelation.
    
    Add StreamingTest coverage against an insecure cluster: requiring
    authentication or encryption must now fail the write instead of
    silently proceeding; both new tests fail without the fix.
    
    Co-Authored-By: Claude Opus 4.8
    
    Change-Id: I661a45fe327d2d8adca5898f211b14233d5d813f
    Reviewed-on: http://gerrit.cloudera.org:8080/24791
    Tested-by: Kudu Jenkins
    Reviewed-by: Gabriella Lotz <lo...@gmail.com>
    Reviewed-by: Zoltan Martonka <zm...@cloudera.com>
---
 .../org/apache/kudu/spark/kudu/DefaultSource.scala |  8 ++-
 .../org/apache/kudu/spark/kudu/StreamingTest.scala | 75 ++++++++++++++++++++++
 2 files changed, 82 insertions(+), 1 deletion(-)

diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
index 6d9c0c996..86a51c56b 100644
--- a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/DefaultSource.scala
@@ -547,7 +547,13 @@ class KuduSink(
     extends Sink {
 
   private val context: KuduContext =
-    new KuduContext(masterAddrs, sqlContext.sparkContext, None, Some(saslProtocolName))
+    new KuduContext(
+      masterAddrs,
+      sqlContext.sparkContext,
+      None,
+      Some(saslProtocolName),
+      requireAuthentication,
+      encryptionPolicy)
 
   override def addBatch(batchId: Long, data: DataFrame): Unit = {
     context.writeRows(data, tableName, operationType, writeOptions)
diff --git a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/StreamingTest.scala b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/StreamingTest.scala
index 1e2eaab2e..ff79d0cae 100644
--- a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/StreamingTest.scala
+++ b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/StreamingTest.scala
@@ -17,10 +17,14 @@
 
 package org.apache.kudu.spark.kudu
 
+import org.apache.kudu.test.KuduTestHarness.MasterServerConfig
+import org.apache.kudu.test.KuduTestHarness.TabletServerConfig
 import org.apache.spark.sql.SQLContext
 import org.apache.spark.sql.execution.streaming._
 import org.apache.spark.sql.streaming.OutputMode
+import org.apache.spark.sql.streaming.StreamingQuery
 import org.junit.Assert.assertEquals
+import org.junit.Assert.assertTrue
 import org.junit.Before
 import org.junit.Test
 
@@ -69,4 +73,75 @@ class StreamingTest extends KuduTestSuite {
     verifyOutput(expectedData = Seq((2, "1"), (3, "2"), (4, "3")))
     query.stop()
   }
+
+  /**
+   * Runs a streaming write to the Kudu table with the given extra options and
+   * asserts that it fails with an error whose (possibly nested) message contains
+   * `expectedMessage`.
+   */
+  private def assertStreamingWriteFails(
+      extraOptions: Map[String, String],
+      expectedMessage: String): Unit = {
+    KuduClientCache.clearCacheForTests()
+    val spark = ss
+    import spark.implicits._
+    val checkpointDir = java.nio.file.Files.createTempDirectory("spark_kudu")
+    val input = MemoryStream[Int]
+    input.addData(1, 2, 3)
+    var query: StreamingQuery = null
+    val exception =
+      try {
+        query = input
+          .toDS()
+          .map(v => (v + 1, v.toString))
+          .toDF("key", "val")
+          .writeStream
+          .format("kudu")
+          .option("kudu.master", harness.getMasterAddressesAsString)
+          .option("kudu.table", simpleTableName)
+          .option("checkpointLocation", checkpointDir.toFile.getCanonicalPath)
+          .options(extraOptions)
+          .outputMode(OutputMode.Update)
+          .start()
+        query.processAllAvailable()
+        None
+      } catch {
+        case e: Throwable => Some(e)
+      } finally {
+        if (query != null) query.stop()
+      }
+    assertTrue("expected the streaming write to fail", exception.isDefined)
+    val messages = Iterator
+      .iterate(exception.get)(_.getCause)
+      .takeWhile(_ != null)
+      .flatMap(e => Option(e.getMessage))
+      .mkString(" | ")
+    assertTrue(s"unexpected exception: $messages", messages.contains(expectedMessage))
+  }
+
+  /**
+   * Verifies that the KuduSink honors the "kudu.requireAuthentication" option.
+   * Against an insecure cluster, requiring authentication must fail the write
+   * rather than silently falling back to the default (no authentication).
+   */
+  @Test
+  def testKuduRequireAuthenticationInsecureClusterStreaming() {
+    assertStreamingWriteFails(
+      Map("kudu.requireAuthentication" -> "true"),
+      "client requires authentication, but server does not have Kerberos enabled")
+  }
+
+  /**
+   * Verifies that the KuduSink honors the "kudu.encryptionPolicy" option.
+   * Against a cluster with RPC encryption disabled, requiring encryption must
+   * fail the write rather than silently falling back to the default (OPTIONAL).
+   */
+  @Test
+  @MasterServerConfig(flags = Array("--rpc_encryption=disabled", "--rpc_authentication=disabled"))
+  @TabletServerConfig(flags = Array("--rpc_encryption=disabled", "--rpc_authentication=disabled"))
+  def testKuduRequireEncryptionInsecureClusterStreaming() {
+    assertStreamingWriteFails(
+      Map("kudu.encryptionPolicy" -> "required_remote"),
+      "server does not support required TLS encryption")
+  }
 }