prakashgyank46-commits opened a new issue, #20325:
URL: https://github.com/apache/hudi/issues/20325
### Bug Description
**What happened:**
`SparkRDDWriteClient.scheduleLogCompaction(Option.empty())` fails with
`NullPointerException: Expected a non-null value. Got null` in
`HoodieCDCDataBlock` when a Merge-on-Read file slice contains CDC log blocks
and the planner takes the block-metadata scanning path. Reproduced with the
official Hudi 1.2.1 Spark 3.5 bundle, and separately with 1.1.0. No Hudi code
was patched or scheduler checks bypassed.
**What you expected:**
Scheduling should inspect valid CDC log blocks without throwing. In this
sparse workload no file group reaches the threshold, so it should return an
empty instant, as the CDC-disabled control does, preserving snapshot/CDC
semantics.
**Steps to reproduce on 1.2.1:**
1. Create a fresh MOR table, CDC enabled (`data_before_after`), partition
`part`, record key `id`, ordering field `ts`. Disable automatic
compaction/cleaning and metadata table.
2. Bulk-insert rows `a`,`b` into partition `p`, then update both in three
separate delta commits.
3. Schedule, execute, and **commit regular compaction**. Verify a physical
`.cdc` log file is generated.
4. Update `a`,`b` in `p` in three more delta commits.
5. Insert/update row `z` in partition `q` across seven separate delta
commits. There are now **10 table-wide delta commits since regular
compaction**, while `p` has only three subsequent data logs plus its CDC log.
6. Set `hoodie.log.compaction.blocks.threshold=10`, construct the write
client with the table schema, call `scheduleLogCompaction(Option.empty())`.
7. The block metadata scanner throws the NPE below.
**Trigger detail:** 1.2.1 has a table-wide delta-commit gate. Three
post-compaction commits with threshold 10 return an early no-op and mask this
failure. Ten commits distributed across file groups pass that gate while
requiring scanning of a sparse file group. A lower threshold can bypass the
scanner through its log-file-count fast path; that does not demonstrate a fix.
**Validation:** 14 local scenarios: 3 successful log compactions, 7 clean
no-ops, 4 expected scheduling NPEs. The four failures cover
`data_before_after`, `data_before`, `op_key_only`, and CDC disabled in table
properties after physical CDC logs were generated. The equivalent distributed
CDC-disabled control returns a clean no-op. Thus existing CDC blocks are the
trigger, not just the current CDC flag. Snapshot rows matched expected values
before the attempt and were unchanged afterward. Full CDC query results
remained unchanged where CDC stayed enabled. No data loss was observed; the
demonstrated impact is failure to schedule log compaction.
The complete standalone Java reproducer, configs, case results and source
analysis are included below. Its `REPRODUCTION_VERIFIED cases=14
expected_NPE=4` result means the bug was reproduced, not fixed.
**Duplicate search:** No matching open or closed report found for
`HoodieCDCDataBlock`, `HoodieLogBlockMetadataScanner`, `logcompaction
NullPointerException`, or `log compaction CDC`.
### Environment
**Hudi version:** 1.2.1, official
`org.apache.hudi:hudi-spark3.5-bundle_2.12:1.2.1`, fresh table version 9. Also
reproduced on 1.1.0; intermediate versions not runtime-tested.
**Query engine:** Spark 3.5.6, Scala 2.12, `local[2]`.
**Runtime:** OpenJDK 11.0.32.1, macOS 26.6/aarch64, local filesystem, 3 GB
heap. No Docker, external catalog, Hive metastore or custom record merger.
**Relevant configs:**
```properties
hoodie.datasource.write.table.type=MERGE_ON_READ
hoodie.datasource.write.recordkey.field=id
hoodie.datasource.write.precombine.field=ts
hoodie.datasource.write.partitionpath.field=part
hoodie.table.cdc.enabled=true
hoodie.table.cdc.supplemental.logging.mode=data_before_after
hoodie.metadata.enable=false
hoodie.compact.inline=false
hoodie.compact.schedule.inline=false
hoodie.compact.inline.max.delta.commits=1
hoodie.clean.automatic=false
hoodie.archive.automatic=false
hoodie.log.compaction.enable=false
hoodie.log.compaction.blocks.threshold=10
hoodie.embed.timeline.server=false
hoodie.write.markers.type=DIRECT
```
Compaction/log compaction are explicitly invoked through the write client.
Shuffle parallelism is 1; Spark uses Kryo and Hudi SQL extension/catalog. The
full harness and options are included below.
The unmodified Maven Central bundle was verified against published SHA-1
`1120807b30891d6e32e4c71512b8f4fbfa2a8d50`.
### Logs and Stack Trace
```text
java.lang.NullPointerException: Expected a non-null value. Got null
at org.apache.hudi.common.util.Option.<init>(Option.java:65)
at org.apache.hudi.common.util.Option.of(Option.java:76)
at
org.apache.hudi.common.table.log.block.HoodieCDCDataBlock.<init>(HoodieCDCDataBlock.java:45)
at
org.apache.hudi.common.table.log.HoodieLogFileReader.readBlock(HoodieLogFileReader.java:212)
at
org.apache.hudi.common.table.log.HoodieLogFileReader.next(HoodieLogFileReader.java:379)
at
org.apache.hudi.common.table.log.HoodieLogFormatReader.next(HoodieLogFormatReader.java:100)
at
org.apache.hudi.common.table.log.BaseHoodieLogRecordReader.scanInternal(BaseHoodieLogRecordReader.java:286)
at
org.apache.hudi.common.table.log.HoodieLogBlockMetadataScanner.<init>(HoodieLogBlockMetadataScanner.java:41)
at
org.apache.hudi.table.action.compact.plan.generators.HoodieLogCompactionPlanGenerator.isFileSliceEligibleForLogCompaction(HoodieLogCompactionPlanGenerator.java:104)
at
org.apache.hudi.table.action.compact.plan.generators.HoodieLogCompactionPlanGenerator.filterFileSlice(HoodieLogCompactionPlanGenerator.java:78)
at
org.apache.hudi.table.action.compact.plan.generators.BaseHoodieCompactionPlanGenerator.lambda$generateCompactionPlan$2(BaseHoodieCompactionPlanGenerator.java:132)
```
**Root-cause analysis:** The metadata scanner creates a reader context
without a schema handler. `BaseHoodieLogRecordReader` therefore has a null
reader schema. `HoodieLogFileReader` passes that null into the CDC block
constructor, which calls `Option.of(readerSchema)`.
Relevant 1.2.1 source:
- [Planner and file-count fast
path](https://github.com/apache/hudi/blob/release-1.2.1/hudi-client/hudi-client-common/src/main/java/org/apache/hudi/table/action/compact/plan/generators/HoodieLogCompactionPlanGenerator.java)
- [Metadata
scanner](https://github.com/apache/hudi/blob/release-1.2.1/hudi-common/src/main/java/org/apache/hudi/common/table/log/HoodieLogBlockMetadataScanner.java)
- [CDC block
constructor](https://github.com/apache/hudi/blob/release-1.2.1/hudi-common/src/main/java/org/apache/hudi/common/table/log/block/HoodieCDCDataBlock.java)
A fix should cover both CDC-enabled tables and retained CDC logs after
disabling CDC, and preserve CDC readability. A null-handling change alone
should not be assumed sufficient without checking the scanner's handling of CDC
block types.
**Duplicate search:** Searched open and closed upstream issues for
`HoodieCDCDataBlock`, `HoodieLogBlockMetadataScanner`, `logcompaction
NullPointerException`, and `log compaction CDC`; no matching report was found.
### Reproducer
The standalone harness below was compiled and run against the official 1.2.1
bundle with the Spark 3.5.6 runtime. Its assertions verify the observed bug and
control outcomes; `REPRODUCTION_VERIFIED cases=14 expected_NPE=4` means the
reproduction succeeded, not that the bug is fixed. The Spark application name
has been generalized for this report.
Save the Java source as `CdcLogCompactionRepro.java`. With Java 11 and
`REPRO_CP` containing the Spark 3.5.6 runtime dependencies plus the Hudi 1.2.1
Spark 3.5/Scala 2.12 bundle (and no older Hudi bundle):
```bash
mkdir -p evidence classes
javac -cp "$REPRO_CP" -d classes CdcLogCompactionRepro.java
SPARK_LOCAL_IP=127.0.0.1 java -Xmx3g -cp "classes:$REPRO_CP"
CdcLogCompactionRepro
```
The test creates fresh disposable tables beneath `evidence/` and retains
them for inspection. It does not connect to a cluster or use any private
dataset.
<details>
<summary>Complete Java reproducer (14 cases)</summary>
```java
import java.nio.file.*;
import java.util.*;
import java.util.stream.*;
import org.apache.spark.sql.*;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.hudi.client.SparkRDDWriteClient;
import org.apache.hudi.client.common.HoodieSparkEngineContext;
import org.apache.hudi.common.table.HoodieTableMetaClient;
import org.apache.hudi.common.table.HoodieTableConfig;
import org.apache.hudi.common.table.TableSchemaResolver;
import org.apache.hudi.common.util.Option;
import org.apache.hudi.config.HoodieWriteConfig;
import org.apache.hudi.hadoop.fs.HadoopFSUtils;
import org.apache.hudi.table.action.HoodieWriteMetadata;
public class CdcLogCompactionRepro {
static SparkSession spark;
static JavaSparkContext jsc;
static Path root;
static int validatedCases;
public static void main(String[] args) throws Exception {
TimeZone.setDefault(TimeZone.getTimeZone("UTC"));
root = Files.createTempDirectory(Paths.get("evidence"),
"tables-").toAbsolutePath();
System.out.println("EVIDENCE_ROOT=" + root);
System.out.println("HUDI_RUNTIME=" +
SparkRDDWriteClient.class.getProtectionDomain().getCodeSource().getLocation());
spark =
SparkSession.builder().master("local[2]").appName("hudi-cdc-logcompaction-repro")
.config("spark.driver.host",
"127.0.0.1").config("spark.driver.bindAddress", "127.0.0.1")
.config("spark.ui.enabled",
"false").config("spark.sql.shuffle.partitions", "1")
.config("spark.default.parallelism", "1").config("spark.serializer",
"org.apache.spark.serializer.KryoSerializer")
.config("spark.sql.extensions",
"org.apache.spark.sql.hudi.HoodieSparkSessionExtension")
.config("spark.sql.catalog.spark_catalog",
"org.apache.spark.sql.hudi.catalog.HoodieCatalog")
.config("spark.sql.adaptive.enabled", "false").getOrCreate();
spark.sparkContext().setLogLevel("ERROR");
jsc = JavaSparkContext.fromSparkContext(spark.sparkContext());
try {
runCase("cdc_disabled", false, false);
runCase("cdc_enabled_logs", true, false);
runCase("cdc_enabled_after_compaction", true, true);
runCase("cdc_disabled_threshold10", false, true);
runCase("cdc_enabled_no_blocks_threshold10", true, false);
runCase("cdc_blocks_threshold10", true, true);
runCase("cdc_data_before_threshold10", true, true);
runCase("cdc_op_key_only_threshold10", true, true);
runCase("cdc_flag_disabled_existing_blocks_threshold10", true, true);
runCase("sparse_cdc_disabled_threshold10", false, true);
runCase("sparse_cdc_blocks_threshold10", true, true);
runCase("sparse_cdc_data_before_threshold10", true, true);
runCase("sparse_cdc_op_key_only_threshold10", true, true);
runCase("sparse_cdc_flag_disabled_existing_blocks_threshold10", true,
true);
if (validatedCases != 14) throw new AssertionError("Missing results: "
+ validatedCases);
System.out.println("REPRODUCTION_VERIFIED cases=" + validatedCases + "
expected_NPE=4");
} finally { spark.stop(); }
}
static Map<String,String> options(String name, boolean cdc) {
Map<String,String> p = new HashMap<>();
p.put("hoodie.table.name", name);
p.put("hoodie.datasource.write.table.type", "MERGE_ON_READ");
p.put("hoodie.datasource.write.recordkey.field", "id");
p.put("hoodie.datasource.write.precombine.field", "ts");
p.put("hoodie.datasource.write.partitionpath.field", "part");
p.put("hoodie.table.cdc.enabled", String.valueOf(cdc));
p.put("hoodie.table.cdc.supplemental.logging.mode", "data_before_after");
p.put("hoodie.metadata.enable", "false");
p.put("hoodie.compact.inline", "false");
p.put("hoodie.compact.schedule.inline", "false");
p.put("hoodie.compact.inline.max.delta.commits", "1");
p.put("hoodie.clean.automatic", "false");
p.put("hoodie.archive.automatic", "false");
p.put("hoodie.log.compaction.enable", "false");
p.put("hoodie.log.compaction.blocks.threshold", "3");
p.put("hoodie.embed.timeline.server", "false");
p.put("hoodie.write.markers.type", "DIRECT");
p.put("hoodie.insert.shuffle.parallelism", "1");
p.put("hoodie.upsert.shuffle.parallelism", "1");
p.put("hoodie.bulkinsert.shuffle.parallelism", "1");
p.put("hoodie.delete.shuffle.parallelism", "1");
return p;
}
static void write(String path, Map<String,String> opts, int n) {
Dataset<Row> df = spark.read().json(spark.createDataset(Arrays.asList(
"{\"id\":\"a\",\"ts\":" + n + ",\"value\":\"a" + n +
"\",\"part\":\"p\"}",
"{\"id\":\"b\",\"ts\":" + n + ",\"value\":\"b" + n +
"\",\"part\":\"p\"}"), Encoders.STRING()));
df.coalesce(1).write().format("hudi").options(opts)
.option("hoodie.datasource.write.operation", n == 0 ? "bulk_insert"
: "upsert")
.mode(SaveMode.Append).save(path);
}
static HoodieTableMetaClient meta(String path) {
return
HoodieTableMetaClient.builder().setConf(HadoopFSUtils.getStorageConf(jsc.hadoopConfiguration()))
.setBasePath(path).setLoadActiveTimelineOnLoad(true).build();
}
static SparkRDDWriteClient client(String path, Map<String,String> opts)
throws Exception {
Properties p = new Properties(); p.putAll(opts);
HoodieWriteConfig cfg =
HoodieWriteConfig.newBuilder().withPath(path).forTable(opts.get("hoodie.table.name"))
.withSchema(new
TableSchemaResolver(meta(path)).getTableSchema().toString()).withProps(p).build();
return new SparkRDDWriteClient(new HoodieSparkEngineContext(jsc), cfg);
}
static List<String> snapshot(String path) {
return
spark.read().format("hudi").load(path).select("id","ts","value","part").orderBy("id").toJSON().collectAsList();
}
static List<String> cdcRows(String path) {
List<String> rows = new ArrayList<>(spark.read().format("hudi")
.option("hoodie.datasource.query.type", "incremental")
.option("hoodie.datasource.query.incremental.format", "cdc")
.option("hoodie.datasource.read.begin.instanttime", "000")
.load(path).toJSON().collectAsList());
Collections.sort(rows);
return rows;
}
static boolean hasNpe(Throwable t) {
for (Throwable c=t;c!=null;c=c.getCause()) {
if (c instanceof NullPointerException &&
c.getMessage().contains("non-null")) return true;
}
return false;
}
static void runCase(String name, boolean cdc, boolean compactFirst) throws
Exception {
System.out.println("CASE_BEGIN=" + name);
String path = root.resolve(name).toString();
Map<String,String> opts = options(name, cdc);
if (name.endsWith("threshold10"))
opts.put("hoodie.log.compaction.blocks.threshold", "10");
if (name.contains("data_before_threshold"))
opts.put("hoodie.table.cdc.supplemental.logging.mode", "data_before");
if (name.contains("op_key_only"))
opts.put("hoodie.table.cdc.supplemental.logging.mode", "op_key_only");
write(path, opts, 0);
for (int n=1;n<=3;n++) write(path, opts, n);
if (compactFirst) {
try (SparkRDDWriteClient c = client(path, opts)) {
Option<String> instant = c.scheduleCompaction(Option.empty());
if (!instant.isPresent()) throw new AssertionError("No regular
compaction scheduled");
HoodieWriteMetadata result = c.compact(instant.get());
c.commitCompaction(instant.get(), result, Option.empty());
System.out.println("REGULAR_COMPACTION=" + instant.get());
}
for (int n=4;n<=6;n++) write(path, opts, n);
}
boolean sparse = name.startsWith("sparse_");
if (sparse) {
for (int n=7;n<=13;n++) {
Dataset<Row> extra =
spark.read().json(spark.createDataset(Collections.singletonList(
"{\"id\":\"z\",\"ts\":" + n + ",\"value\":\"z" + n +
"\",\"part\":\"q\"}"), Encoders.STRING()));
extra.coalesce(1).write().format("hudi").options(opts)
.option("hoodie.datasource.write.operation",
"upsert").mode(SaveMode.Append).save(path);
}
System.out.println("SPARSE_FIXTURE=10 delta commits after regular
compaction; only 3 update partition p");
}
List<String> before = snapshot(path);
int expectedTs = compactFirst ? 6 : 3;
List<String> expected = Arrays.asList(
"{\"id\":\"a\",\"ts\":" + expectedTs + ",\"value\":\"a" + expectedTs
+ "\",\"part\":\"p\"}",
"{\"id\":\"b\",\"ts\":" + expectedTs + ",\"value\":\"b" + expectedTs
+ "\",\"part\":\"p\"}");
if (sparse) {
expected = new ArrayList<>(expected);
expected.add("{\"id\":\"z\",\"ts\":13,\"value\":\"z13\",\"part\":\"q\"}");
}
if (!before.equals(expected)) throw new AssertionError("Unexpected
fixture snapshot " + before);
boolean disableCdcFlag = name.contains("flag_disabled");
List<String> cdcBefore = cdc && !disableCdcFlag ? cdcRows(path) : null;
if (cdcBefore != null) System.out.println("CDC_ROWS_BEFORE=" +
cdcBefore.size());
if (disableCdcFlag) {
HoodieTableMetaClient m = meta(path);
Properties changed = new Properties();
changed.setProperty("hoodie.table.cdc.enabled", "false");
HoodieTableConfig.update(m.getStorage(), m.getMetaPath(), changed);
opts.put("hoodie.table.cdc.enabled", "false");
if (meta(path).getTableConfig().isCDCEnabled()) throw new
AssertionError("CDC flag still enabled");
System.out.println("CDC_FLAG_DISABLED retaining existing CDC log");
}
boolean expectNpe = sparse && cdc;
long cdcFiles;
System.out.println("SNAPSHOT_BEFORE=" + before);
try (Stream<Path> paths = Files.walk(Paths.get(path))) {
List<Path> logs = paths.filter(p ->
p.getFileName().toString().contains(".log.") &&
!p.toString().endsWith(".crc")).collect(Collectors.toList());
cdcFiles = logs.stream().filter(p ->
p.toString().endsWith(".cdc")).count();
logs.forEach(p -> System.out.println("LOG_FILE=" + p));
}
if ((cdc && compactFirst) != (cdcFiles > 0)) throw new
AssertionError("Unexpected CDC file count " + cdcFiles);
String stage = "schedule";
try (SparkRDDWriteClient c = client(path, opts)) {
Option<String> instant = c.scheduleLogCompaction(Option.empty());
if (!instant.isPresent()) {
if (expectNpe || !name.endsWith("threshold10")) throw new
AssertionError("Unexpected unscheduled log compaction");
if (!before.equals(snapshot(path))) throw new
AssertionError("Snapshot changed");
if (cdcBefore != null && !cdcBefore.equals(cdcRows(path))) throw new
AssertionError("CDC changed");
System.out.println("CASE_RESULT=" + name + " NO_OP below threshold;
snapshot unchanged=" + before.equals(snapshot(path)));
validatedCases++;
return;
}
System.out.println("LOG_COMPACTION_SCHEDULED=" + instant.get());
stage = "execute";
HoodieWriteMetadata result = c.logCompact(instant.get());
c.commitLogCompaction(instant.get(), result, Option.empty());
stage = "verify";
List<String> after = snapshot(path);
if (!before.equals(after)) throw new AssertionError("Snapshot changed:
" + after);
if (expectNpe) throw new AssertionError("Expected reproduction
missing");
if (cdcBefore != null && !cdcBefore.equals(cdcRows(path))) throw new
AssertionError("CDC changed after log compaction");
System.out.println("CASE_RESULT=" + name + " PASS snapshot unchanged
rows=" + after.size());
validatedCases++;
} catch (Throwable t) {
System.out.println("CASE_RESULT=" + name + " FAIL stage=" + stage);
t.printStackTrace(System.out);
List<String> after = snapshot(path);
System.out.println("SNAPSHOT_AFTER_FAILURE=" + after);
if (!expectNpe || !stage.equals("schedule") || !hasNpe(t)) throw new
AssertionError("Unexpected failure", t);
if (!before.equals(after)) throw new AssertionError("Snapshot changed
after failure");
if (cdcBefore != null && !cdcBefore.equals(cdcRows(path))) throw new
AssertionError("CDC changed after failure");
System.out.println("EXPECTED_NPE_VERIFIED=" + name + "
snapshot_and_available_CDC_unchanged");
validatedCases++;
}
}
}
```
</details>
<details>
<summary>Observed case results</summary>
```text
CASE_RESULT=cdc_disabled PASS snapshot unchanged rows=2
CASE_RESULT=cdc_enabled_logs PASS snapshot unchanged rows=2
CASE_RESULT=cdc_enabled_after_compaction PASS snapshot unchanged rows=2
CASE_RESULT=cdc_disabled_threshold10 NO_OP below threshold; snapshot
unchanged=true
CASE_RESULT=cdc_enabled_no_blocks_threshold10 NO_OP below threshold;
snapshot unchanged=true
CASE_RESULT=cdc_blocks_threshold10 NO_OP below threshold; snapshot
unchanged=true
CASE_RESULT=cdc_data_before_threshold10 NO_OP below threshold; snapshot
unchanged=true
CASE_RESULT=cdc_op_key_only_threshold10 NO_OP below threshold; snapshot
unchanged=true
CASE_RESULT=cdc_flag_disabled_existing_blocks_threshold10 NO_OP below
threshold; snapshot unchanged=true
CASE_RESULT=sparse_cdc_disabled_threshold10 NO_OP below threshold; snapshot
unchanged=true
CASE_RESULT=sparse_cdc_blocks_threshold10 FAIL stage=schedule
EXPECTED_NPE_VERIFIED=sparse_cdc_blocks_threshold10
snapshot_and_available_CDC_unchanged
CASE_RESULT=sparse_cdc_data_before_threshold10 FAIL stage=schedule
EXPECTED_NPE_VERIFIED=sparse_cdc_data_before_threshold10
snapshot_and_available_CDC_unchanged
CASE_RESULT=sparse_cdc_op_key_only_threshold10 FAIL stage=schedule
EXPECTED_NPE_VERIFIED=sparse_cdc_op_key_only_threshold10
snapshot_and_available_CDC_unchanged
CASE_RESULT=sparse_cdc_flag_disabled_existing_blocks_threshold10 FAIL
stage=schedule
EXPECTED_NPE_VERIFIED=sparse_cdc_flag_disabled_existing_blocks_threshold10
snapshot_and_available_CDC_unchanged
REPRODUCTION_VERIFIED cases=14 expected_NPE=4
```
</details>
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
