This is an automated email from the ASF dual-hosted git repository.
danny0405 pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/hudi.git
The following commit(s) were added to refs/heads/master by this push:
new 76a95f5d629b fix(utilities): guard Source.releaseResources() against
transient RDD unpersist failures (#19328)
76a95f5d629b is described below
commit 76a95f5d629bceff89eab081c768438f2fef1e87
Author: Nikhil Kumar <[email protected]>
AuthorDate: Wed Jul 22 15:47:46 2026 +0530
fix(utilities): guard Source.releaseResources() against transient RDD
unpersist failures (#19328)
releaseResources() runs in StreamSync.syncOnce()'s finally block, after the
write/commit has already completed. A transient Spark NullPointerException
in
BlockManagerMaster.removeRdd (thrown when the SparkContext is mid-teardown
or
stopping and the driver endpoint is null) during the post-write RDD
unpersist
was propagating out and failing otherwise-successful ingestion rounds. A
cleanup-only failure must not fail the round: wrap the unpersist in
try/catch
and log at WARN. Extract the unpersist into a protected
unpersistCachedSourceRdd()
seam (VisibleForTesting) and add TestSource covering both the swallow and
happy paths.
---
.../org/apache/hudi/utilities/sources/Source.java | 12 ++++
.../apache/hudi/utilities/sources/TestSource.java | 73 ++++++++++++++++++++++
2 files changed, 85 insertions(+)
diff --git
a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
index 75cef40c1ee5..de2ce192401e 100644
--- a/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
+++ b/hudi-utilities/src/main/java/org/apache/hudi/utilities/sources/Source.java
@@ -28,6 +28,7 @@ import
org.apache.hudi.common.table.checkpoint.StreamerCheckpointV2;
import org.apache.hudi.common.util.ConfigUtils;
import org.apache.hudi.common.util.Either;
import org.apache.hudi.common.util.Option;
+import org.apache.hudi.common.util.VisibleForTesting;
import org.apache.hudi.utilities.callback.SourceCommitCallback;
import org.apache.hudi.utilities.schema.SchemaProvider;
import org.apache.hudi.utilities.streamer.DefaultStreamContext;
@@ -172,6 +173,17 @@ public abstract class Source<T> implements
SourceCommitCallback, Serializable {
@Override
public void releaseResources() {
+ // Cleanup runs after the write/commit; a transient Spark failure while
unpersisting
+ // must not fail an already-successful round.
+ try {
+ unpersistCachedSourceRdd();
+ } catch (Exception e) {
+ log.warn("Failed to unpersist cached source RDD during releaseResources;
ignoring", e);
+ }
+ }
+
+ @VisibleForTesting
+ protected void unpersistCachedSourceRdd() {
if (cachedSourceRdd != null && cachedSourceRdd.isLeft()) {
cachedSourceRdd.asLeft().unpersist();
} else if (cachedSourceRdd != null && cachedSourceRdd.isRight()) {
diff --git
a/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
new file mode 100644
index 000000000000..e6e4897e05f1
--- /dev/null
+++
b/hudi-utilities/src/test/java/org/apache/hudi/utilities/sources/TestSource.java
@@ -0,0 +1,73 @@
+/*
+ * 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.hudi.utilities.sources;
+
+import org.apache.hudi.common.config.TypedProperties;
+import org.apache.hudi.common.util.Option;
+
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
+/**
+ * Unit tests for {@link Source#releaseResources()}.
+ *
+ * <p>releaseResources() is invoked from StreamSync.syncOnce()'s finally
block, after the
+ * write/commit has already completed. A transient Spark failure while
unpersisting the
+ * cached source RDD (e.g. a NullPointerException from
BlockManagerMaster.removeRdd when
+ * the SparkContext is mid-teardown/stopping) must not fail an
otherwise-successful
+ * ingestion round.
+ */
+public class TestSource {
+
+ /**
+ * Minimal concrete {@link Source} whose unpersist step always fails, so the
+ * swallow-on-failure behaviour of releaseResources() can be exercised
without a
+ * live SparkContext.
+ */
+ private static class MockSourceWithUnpersistenceFailure extends
Source<String> {
+ private int unpersistCalls = 0;
+
+ MockSourceWithUnpersistenceFailure(TypedProperties props) {
+ super(props, null, null, null);
+ }
+
+ @Override
+ protected InputBatch<String> fetchNewData(Option<String> lastCkptStr, long
sourceLimit) {
+ // Not exercised by these tests; releaseResources() is tested directly.
+ return null;
+ }
+
+ @Override
+ protected void unpersistCachedSourceRdd() {
+ unpersistCalls++;
+ throw new NullPointerException("simulated BlockManagerMaster.removeRdd
NPE");
+ }
+ }
+
+ @Test
+ public void releaseResourcesSwallowsTransientUnpersistFailure() {
+ MockSourceWithUnpersistenceFailure source =
+ new MockSourceWithUnpersistenceFailure(new TypedProperties());
+ assertDoesNotThrow(source::releaseResources,
+ "A transient unpersist failure in cleanup must not propagate out of
releaseResources()");
+ assertEquals(1, source.unpersistCalls, "unpersist should have been
attempted exactly once");
+ }
+}
