You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kudu.apache.org by zc...@apache.org on 2026/07/02 07:28:40 UTC
(kudu) branch master updated (8e54d7be0 -> bb85047ed)
This is an automated email from the ASF dual-hosted git repository.
zchovan pushed a change to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
from 8e54d7be0 KUDU-3787 Make tls_debug support Kerberos
new 3a4068a26 KUDU-3790: Fix NPE in HdrHistogramAccumulator.isZero()
new bb85047ed [java] Run example tests on JDK 17
The 2 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
examples/java/java-example/pom.xml | 10 ++
.../kudu/spark/kudu/HdrHistogramAccumulator.scala | 55 +++++++--
.../spark/kudu/HdrHistogramAccumulatorTest.scala | 135 +++++++++++++++++++++
3 files changed, 188 insertions(+), 12 deletions(-)
create mode 100644 java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulatorTest.scala
(kudu) 02/02: [java] Run example tests on JDK 17
Posted by zc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zchovan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit bb85047edf6d1c39474ed940ced33ecf64e22c45
Author: Zoltan Chovan <zc...@cloudera.com>
AuthorDate: Tue Jun 23 17:56:21 2026 +0300
[java] Run example tests on JDK 17
The java-example tests rely on reflective access that JDK 17's strong
module encapsulation blocks by default. Add the maven-surefire-plugin
(3.2.5) and pass --add-opens for java.base/java.lang and
java.base/java.net so the example's tests run on JDK 17.
Change-Id: I4d85afcc8707af3e8c6397ad4267d78470d6df66
Reviewed-on: http://gerrit.cloudera.org:8080/24519
Tested-by: Kudu Jenkins
Reviewed-by: Marton Greber <gr...@gmail.com>
Reviewed-by: Zoltan Martonka <zm...@cloudera.com>
---
examples/java/java-example/pom.xml | 10 ++++++++++
1 file changed, 10 insertions(+)
diff --git a/examples/java/java-example/pom.xml b/examples/java/java-example/pom.xml
index 4775bef03..f9a23bb3d 100644
--- a/examples/java/java-example/pom.xml
+++ b/examples/java/java-example/pom.xml
@@ -29,6 +29,8 @@
<properties>
<kudu-version>1.18.1</kudu-version>
+ <argLine></argLine>
+ <kudu.test.addOpens>--add-opens=java.base/java.lang=ALL-UNNAMED --add-opens=java.base/java.net=ALL-UNNAMED</kudu.test.addOpens>
</properties>
<build>
@@ -42,6 +44,14 @@
<target>1.8</target>
</configuration>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-surefire-plugin</artifactId>
+ <version>3.2.5</version>
+ <configuration>
+ <argLine>${argLine} ${kudu.test.addOpens}</argLine>
+ </configuration>
+ </plugin>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
(kudu) 01/02: KUDU-3790: Fix NPE in HdrHistogramAccumulator.isZero()
Posted by zc...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
zchovan pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/kudu.git
commit 3a4068a26bd919021345b4adf2e89ab7d6da531b
Author: zchovan <zc...@cloudera.com>
AuthorDate: Tue Jun 30 18:06:36 2026 +0200
KUDU-3790: Fix NPE in HdrHistogramAccumulator.isZero()
Spark's executor-heartbeater thread periodically calls isZero() on every
running task's accumulators. The executor-side copy of KuduContext's
"kudu.write_duration" accumulator can have its internal histogram field
observed as null before the task records its first write (unsafe
publication across threads, or constructor-bypassing deserialization such
as Kryo). isZero() then threw a NullPointerException on the heartbeater
thread, which Spark treats as fatal, killing the executor.
In this change, isZero() (and toString()) was made null-tolerant: a
null/absent histogram is logically zero.
isZero() is also now lock-free so it cannot depend on any field being
published.
Additionally HistogramWrapper now synchronizes its mutators on `this`
rather than on the reassignable `innerHistogram` var, which previously
changed the monitor object over time and provided no real mutual exclusion.
Adds HdrHistogramAccumulatorTest covering the zero semantics, a
serialization round-trip, and regression tests that force a null field
and assert isZero() does not throw.
Change-Id: Ia3d556aaf1051464303f35281d1d698b20d5ef20
Reviewed-on: http://gerrit.cloudera.org:8080/24527
Reviewed-by: Marton Greber <gr...@gmail.com>
Tested-by: Marton Greber <gr...@gmail.com>
Reviewed-by: Zoltan Martonka <zm...@cloudera.com>
---
.../kudu/spark/kudu/HdrHistogramAccumulator.scala | 55 +++++++--
.../spark/kudu/HdrHistogramAccumulatorTest.scala | 135 +++++++++++++++++++++
2 files changed, 178 insertions(+), 12 deletions(-)
diff --git a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulator.scala b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulator.scala
index c775930c2..8887b2254 100644
--- a/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulator.scala
+++ b/java/kudu-spark/src/main/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulator.scala
@@ -40,9 +40,22 @@ private[kudu] class HdrHistogramAccumulator(histogram: HistogramWrapper = new Hi
extends AccumulatorV2[Int, HistogramWrapper] {
override def isZero: Boolean = {
- histogram.isZero
+ // Read the field into a local. This method is invoked by Spark's
+ // heartbeater thread on freshly-deserialized copies of the accumulator and
+ // must not throw if `histogram` is observed null -- e.g. due to unsafe
+ // publication across threads before the field write is visible, or due to
+ // constructor-bypassing deserialization (Kryo). A null/absent histogram is
+ // logically zero.
+ val h = histogram
+ h == null || h.isZero
}
+ // copy(), reset(), add(), merge() and value below intentionally do NOT guard
+ // against a null `histogram`. Unlike isZero()/toString() -- which Spark's
+ // heartbeater thread invokes on accumulator copies it did not construct --
+ // these run only on the task-owning executor thread or the driver, both of
+ // which have a happens-before relationship with construction/deserialization
+ // and therefore always observe a fully-published, non-null histogram.
override def copy(): AccumulatorV2[Int, HistogramWrapper] = {
new HdrHistogramAccumulator(histogram.copy())
}
@@ -61,7 +74,10 @@ private[kudu] class HdrHistogramAccumulator(histogram: HistogramWrapper = new Hi
override def value: HistogramWrapper = histogram
- override def toString: String = histogram.toString
+ override def toString: String = {
+ val h = histogram
+ if (h == null) "0ms" else h.toString
+ }
}
/*
@@ -73,23 +89,38 @@ private[kudu] class HdrHistogramAccumulator(histogram: HistogramWrapper = new Hi
*
* An option is used for innerHistogram so we can only initialize the histogram if it is used.
*/
-private[kudu] class HistogramWrapper(var innerHistogram: Option[IntCountsHistogram] = None)
+private[kudu] class HistogramWrapper(
+ @volatile var innerHistogram: Option[IntCountsHistogram] = None)
extends Serializable {
+ // Mutations of `innerHistogram` synchronize on `this`. The previous code
+ // synchronized on the `innerHistogram` field itself, but that field is a
+ // reassignable var, so the monitor changed over time (None vs each Some(...))
+ // and did not actually provide mutual exclusion. `this` is a stable monitor
+ // and -- unlike a dedicated lock object -- does not break serialization
+ // (HistogramWrapper is Serializable and is shipped with the Spark task).
+ //
+ // Note: isZero deliberately does NOT synchronize. It is called by Spark's
+ // heartbeater thread on freshly-deserialized accumulator copies and must
+ // remain safe (and lock-free) even if instance fields are observed null.
+ // `innerHistogram` is @volatile so this lock-free read has a JMM visibility
+ // guarantee on weak memory models; NPE-safety still comes from the null
+ // check, since the pre-publication default is observable regardless.
def isZero: Boolean = {
- innerHistogram.synchronized {
- innerHistogram.isEmpty
- }
+ // A single atomic reference read; checking emptiness does not touch the
+ // (mutable) underlying histogram, so no locking is required here.
+ val h = innerHistogram
+ h == null || h.isEmpty
}
def copy(): HistogramWrapper = {
- innerHistogram.synchronized {
+ this.synchronized {
new HistogramWrapper(innerHistogram.map(_.copy()))
}
}
def reset(): Unit = {
- innerHistogram.synchronized {
+ this.synchronized {
if (innerHistogram.isDefined) {
innerHistogram.get.reset()
}
@@ -98,14 +129,14 @@ private[kudu] class HistogramWrapper(var innerHistogram: Option[IntCountsHistogr
}
def add(v: Int) {
- innerHistogram.synchronized {
+ this.synchronized {
initializeIfEmpty()
innerHistogram.get.recordValue(v)
}
}
def add(other: HistogramWrapper) {
- innerHistogram.synchronized {
+ this.synchronized {
if (other.innerHistogram.isEmpty) {
return
}
@@ -121,8 +152,8 @@ private[kudu] class HistogramWrapper(var innerHistogram: Option[IntCountsHistogr
}
override def toString: String = {
- innerHistogram.synchronized {
- if (innerHistogram.isEmpty) {
+ this.synchronized {
+ if (innerHistogram == null || innerHistogram.isEmpty) {
return "0ms"
}
diff --git a/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulatorTest.scala b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulatorTest.scala
new file mode 100644
index 000000000..43c740497
--- /dev/null
+++ b/java/kudu-spark/src/test/scala/org/apache/kudu/spark/kudu/HdrHistogramAccumulatorTest.scala
@@ -0,0 +1,135 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+package org.apache.kudu.spark.kudu
+
+import java.io.ByteArrayInputStream
+import java.io.ByteArrayOutputStream
+import java.io.ObjectInputStream
+import java.io.ObjectOutputStream
+
+import org.junit.Test
+import org.scalatest.matchers.should.Matchers
+
+/**
+ * Unit tests for [[HdrHistogramAccumulator]] and [[HistogramWrapper]].
+ *
+ * These do not require a Kudu cluster: they exercise the accumulator's
+ * value/zero semantics directly.
+ */
+class HdrHistogramAccumulatorTest extends Matchers {
+
+ @Test
+ def testIsZeroOnFreshAccumulator(): Unit = {
+ val acc = new HdrHistogramAccumulator()
+ acc.isZero shouldBe true
+ acc.toString shouldBe "0ms"
+ }
+
+ @Test
+ def testIsZeroAfterAdd(): Unit = {
+ val acc = new HdrHistogramAccumulator()
+ acc.add(5)
+ acc.isZero shouldBe false
+ }
+
+ @Test
+ def testCopyAndResetIsZero(): Unit = {
+ val acc = new HdrHistogramAccumulator()
+ acc.add(42)
+ acc.isZero shouldBe false
+ // Spark calls copyAndReset() when shipping the accumulator to executors;
+ // the returned copy must be zero (Spark also asserts this internally).
+ val fresh = acc.copyAndReset()
+ fresh.isZero shouldBe true
+ }
+
+ @Test
+ def testResetMakesZero(): Unit = {
+ val acc = new HdrHistogramAccumulator()
+ acc.add(7)
+ acc.reset()
+ acc.isZero shouldBe true
+ }
+
+ /**
+ * Regression test: Spark's executor-heartbeater thread calls isZero() on the
+ * accumulators of running tasks. The executor-side copy can have its internal
+ * `histogram` field observed as null -- e.g. due to unsafe publication across
+ * threads before the field write is visible, or constructor-bypassing
+ * deserialization (Kryo) -- before the task records its first write. isZero()
+ * must treat that as "zero" rather than throwing a NullPointerException,
+ * because an uncaught exception on the heartbeater thread kills the executor.
+ */
+ @Test
+ def testIsZeroWithNullHistogramDoesNotThrow(): Unit = {
+ val acc = new HdrHistogramAccumulator()
+ setFieldToNull(acc, classOf[HistogramWrapper])
+ acc.isZero shouldBe true
+ acc.toString shouldBe "0ms"
+ }
+
+ /** As above, but with the wrapper's inner Option observed as null. */
+ @Test
+ def testIsZeroWithNullInnerHistogramDoesNotThrow(): Unit = {
+ val wrapper = new HistogramWrapper()
+ setFieldToNull(wrapper, classOf[Option[_]])
+ wrapper.isZero shouldBe true
+ wrapper.toString shouldBe "0ms"
+ }
+
+ /**
+ * Plain Java serialization of the wrapper (it is Serializable) must round-trip
+ * to a valid, empty (zero) state.
+ */
+ @Test
+ def testWrapperSerializationRoundTripIsZero(): Unit = {
+ val wrapper = new HistogramWrapper()
+ val restored = deserialize(serialize(wrapper)).asInstanceOf[HistogramWrapper]
+ restored.isZero shouldBe true
+ }
+
+ /** Sets the first declared field of the given type on `target` to null. */
+ private def setFieldToNull(target: AnyRef, fieldType: Class[_]): Unit = {
+ val field = target.getClass.getDeclaredFields
+ .find(f => fieldType.isAssignableFrom(f.getType))
+ .getOrElse(throw new AssertionError(
+ s"no field of type ${fieldType.getName} on ${target.getClass.getName}"))
+ field.setAccessible(true)
+ field.set(target, null)
+ }
+
+ private def serialize(value: Any): Array[Byte] = {
+ val stream = new ByteArrayOutputStream()
+ val oos = new ObjectOutputStream(stream)
+ try {
+ oos.writeObject(value)
+ stream.toByteArray
+ } finally {
+ oos.close()
+ }
+ }
+
+ private def deserialize(bytes: Array[Byte]): Any = {
+ val ois = new ObjectInputStream(new ByteArrayInputStream(bytes))
+ try {
+ ois.readObject
+ } finally {
+ ois.close()
+ }
+ }
+}