This is an automated email from the ASF dual-hosted git repository.
morningman pushed a commit to branch branch-4.0
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.0 by this push:
new f8ba3c2428c [branch-4.0][fix](auto-partition) avoid NPE in
createPartition when a partition is concurrently dropped (#65357)
f8ba3c2428c is described below
commit f8ba3c2428c2d58bf49191659378f0b677ac6d4e
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Wed Jul 8 20:21:08 2026 +0800
[branch-4.0][fix](auto-partition) avoid NPE in createPartition when a
partition is concurrently dropped (#65357)
## Proposed changes
Fix two flaky failures found in a branch-4.0 P0 regression run (both in
the backup/restore area).
Root causes were confirmed from the FE/BE logs of that run.
### 1. `test_backup_restore_retention_count` — NPE in `createPartition`
under concurrent partition drop
An auto-partition INSERT creates partitions on the fly via the
`createPartition` RPC.
`FrontendServiceImpl.createPartition()` adds the partitions and then
reads them back to build the
tablet-location response, **without null-checking**
`table.getPartition(partitionName)`. When
`DynamicPartitionScheduler` concurrently drops a just-created history
partition (enforcing
`partition.retention_count`), the read-back gets `null` and
`partition.getId()` throws NPE:
```
Caused by: java.lang.NullPointerException: Cannot invoke
"org.apache.doris.catalog.Partition.getId()" because "partition" is null
at
org.apache.doris.service.FrontendServiceImpl.createPartition(FrontendServiceImpl.java:3885)
```
The unchecked exception escapes to the Thrift layer as `Internal error
processing createPartition`,
failing the load with `THRIFT_RPC_ERROR`. This can affect any
auto-partition table with
dynamic-partition / `retention_count` cleanup, not only the test.
- FE: null-check the read-back and return a retryable error status
instead of throwing NPE.
- Test: retry the INSERT on the transient race (surfaced when a
concurrent suite lowers the global
`dynamic_partition_check_interval_seconds`).
### 2. `test_ddl_backup_auth` — global backup lock contention
BACKUP submission serializes on a single global `BackupHandler` lock via
`tryLock(10s)`. Under
parallel P0 with many concurrent backup/restore suites, that lock can be
held past 10s by slow S3
operations, so submission transiently fails with `Another backup or
restore job is being submitted`.
The suite only verifies auth, so it now retries the submit on that
contention.
---
.../insert/streaming/StreamingInsertJob.java | 43 ++++++++++++++++++----
.../apache/doris/service/FrontendServiceImpl.java | 12 +++++-
.../suites/auth_call/test_ddl_backup_auth.groovy | 24 ++++++++++--
.../test_backup_restore_retention_count.groovy | 29 ++++++++++++---
4 files changed, 90 insertions(+), 18 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
index 91f193d098d..72c6e1725e2 100644
---
a/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
+++
b/fe/fe-core/src/main/java/org/apache/doris/job/extensions/insert/streaming/StreamingInsertJob.java
@@ -416,6 +416,27 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
this.setJobRuntimeMsg("");
}
+ /**
+ * Clear the failure info after a task succeeds, but only if the job has
not been paused.
+ * A streaming job runs the data task and the meta-fetch scheduler task
concurrently on
+ * different thread pools. If a fetchMeta failure has paused (or is
pausing) the job, a
+ * straggler successful task must not wipe out the pause reason via
resetFailureInfo(null):
+ * doing so hides the cause (empty ErrorMsg) and, because auto resume only
fires when
+ * failureReason != null, leaves the job stuck in PAUSED forever. Guarding
under the write
+ * lock (which the fetchMeta pause path also takes) makes the two paths
mutually exclusive
+ * so the pause reason always survives.
+ */
+ private void clearFailureInfoIfNotPaused() {
+ writeLock();
+ try {
+ if (!JobStatus.PAUSED.equals(getJobStatus())) {
+ resetFailureInfo(null);
+ }
+ } finally {
+ writeUnlock();
+ }
+ }
+
@Override
public void cancelAllTasks(boolean needWaitCancelComplete) throws
JobException {
lock.writeLock().lock();
@@ -543,12 +564,20 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
||
!InternalErrorCode.MANUAL_PAUSE_ERR.equals(this.getFailureReason().getCode())) {
// When a job is manually paused, it does not need to be set
again,
// otherwise, it may be woken up by auto resume.
- this.setFailureReason(
- new
FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
- "Failed to fetch meta, " + ex.getMessage()));
- // If fetching meta fails, the job is paused
- // and auto resume will automatically wake it up.
- this.updateJobStatus(JobStatus.PAUSED);
+ // Set the failure reason and pause atomically under the write
lock so a
+ // concurrent successful task (onStreamTaskSuccess) cannot
clear the reason
+ // between these two writes. See clearFailureInfoIfNotPaused().
+ writeLock();
+ try {
+ this.setFailureReason(
+ new
FailureReason(InternalErrorCode.GET_REMOTE_DATA_ERROR,
+ "Failed to fetch meta, " +
ex.getMessage()));
+ // If fetching meta fails, the job is paused
+ // and auto resume will automatically wake it up.
+ this.updateJobStatus(JobStatus.PAUSED);
+ } finally {
+ writeUnlock();
+ }
if (MetricRepo.isInit) {
MetricRepo.COUNTER_STREAMING_JOB_GET_META_FAIL_COUNT.increase(1L);
@@ -617,7 +646,7 @@ public class StreamingInsertJob extends
AbstractJob<StreamingJobSchedulerTask, M
public void onStreamTaskSuccess(AbstractStreamingTask task) {
try {
- resetFailureInfo(null);
+ clearFailureInfoIfNotPaused();
succeedTaskCount.incrementAndGet();
//update metric
if (MetricRepo.isInit) {
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
index f39367b0e45..12390e1952e 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/service/FrontendServiceImpl.java
@@ -3877,8 +3877,18 @@ public class FrontendServiceImpl implements
FrontendService.Iface {
String cachedClusterId = null;
for (String partitionName : addPartitionClauseMap.keySet()) {
Partition partition = table.getPartition(partitionName);
+ if (partition == null) {
+ // The partition was just created above, but may have been
concurrently dropped before this
+ // read-back, e.g. by DynamicPartitionScheduler enforcing
partition.retention_count or dynamic
+ // partition cleanup. Return a retryable error status instead
of throwing NPE on partition.getId().
+ errorStatus.setErrorMsgs(Lists.newArrayList(String.format(
+ "partition %s was concurrently dropped, please retry",
partitionName)));
+ result.setStatus(errorStatus);
+ LOG.warn("send create partition error status: {}", result);
+ return result;
+ }
// For thread safety, we preserve the tablet distribution
information of each partition
- // before calling getOrSetAutoPartitionInfo, but not check the
partition first
+ // before calling getOrSetAutoPartitionInfo.
List<TTabletLocation> partitionTablets = new ArrayList<>();
List<TTabletLocation> partitionSlaveTablets = new ArrayList<>();
TOlapTablePartition tPartition = new TOlapTablePartition();
diff --git a/regression-test/suites/auth_call/test_ddl_backup_auth.groovy
b/regression-test/suites/auth_call/test_ddl_backup_auth.groovy
index 1a2f7b79718..4c00d5bae18 100644
--- a/regression-test/suites/auth_call/test_ddl_backup_auth.groovy
+++ b/regression-test/suites/auth_call/test_ddl_backup_auth.groovy
@@ -109,10 +109,26 @@ suite("test_ddl_backup_auth","p0,auth_call") {
assertTrue(res.size() == 0)
connect(user, "${pwd}", context.config.jdbcUrl) {
- sql """BACKUP SNAPSHOT ${dbName}.${backupLabelName}
- TO ${repositoryName}
- ON (${tableName})
- PROPERTIES ("type" = "full");"""
+ // BACKUP submission serializes on a single global BackupHandler lock
(tryLock(10s)). Under parallel
+ // P0 with many concurrent backup/restore suites, that lock can be
held past 10s by slow S3 ops,
+ // so submission may transiently fail with "Another backup or restore
job is being submitted".
+ // This suite only verifies auth (a user with LOAD_PRIV can submit
backup), so retry the contention.
+ def maxBackupRetries = 15
+ for (int i = 0; i < maxBackupRetries; i++) {
+ try {
+ sql """BACKUP SNAPSHOT ${dbName}.${backupLabelName}
+ TO ${repositoryName}
+ ON (${tableName})
+ PROPERTIES ("type" = "full");"""
+ break
+ } catch (Exception e) {
+ if (i == maxBackupRetries - 1 ||
!e.getMessage().contains("Another backup or restore job")) {
+ throw e
+ }
+ logger.warn("backup submit contended on global lock, retry ${i
+ 1}: ${e.getMessage()}")
+ sleep(2000)
+ }
+ }
res = sql """SHOW BACKUP FROM ${dbName};"""
logger.info("res: " + res)
assertTrue(res.size() == 1)
diff --git
a/regression-test/suites/backup_restore/test_backup_restore_retention_count.groovy
b/regression-test/suites/backup_restore/test_backup_restore_retention_count.groovy
index 96fe217eac8..45914e61aaa 100644
---
a/regression-test/suites/backup_restore/test_backup_restore_retention_count.groovy
+++
b/regression-test/suites/backup_restore/test_backup_restore_retention_count.groovy
@@ -39,12 +39,29 @@ suite("test_backup_restore_retention_count",
"backup_restore") {
)
"""
- // Insert data to create partitions
- sql """
- INSERT INTO ${dbName}.${tableName}
- SELECT date_add('2020-01-01 00:00:00', interval number day)
- FROM numbers("number" = "10")
- """
+ // Insert data to create partitions.
+ // Auto-partition on-the-fly createPartition can race with
DynamicPartitionScheduler's retention
+ // cleanup (partition.retention_count) when a concurrent suite lowers the
global
+ // dynamic_partition_check_interval_seconds: the scheduler may drop a
just-created history partition
+ // before the createPartition RPC finishes, surfacing as a transient
createPartition RPC error.
+ // Retry the insert on that transient contention.
+ def maxInsertRetries = 10
+ for (int i = 0; i < maxInsertRetries; i++) {
+ try {
+ sql """
+ INSERT INTO ${dbName}.${tableName}
+ SELECT date_add('2020-01-01 00:00:00', interval number day)
+ FROM numbers("number" = "10")
+ """
+ break
+ } catch (Exception e) {
+ if (i == maxInsertRetries - 1 ||
!e.getMessage().contains("createPartition")) {
+ throw e
+ }
+ logger.warn("insert raced with retention cleanup, retry ${i + 1}:
${e.getMessage()}")
+ sleep(2000)
+ }
+ }
sql "alter table ${dbName}.${tableName} set ('partition.retention_count'
= '4')"
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
