This is an automated email from the ASF dual-hosted git repository.
yiguolei pushed a commit to branch branch-4.1
in repository https://gitbox.apache.org/repos/asf/doris.git
The following commit(s) were added to refs/heads/branch-4.1 by this push:
new a806cc2bc43 branch-4.1: [fix](load)(regression-test) fix broker load
label self-conflict on pending retry; deflake P2 suites #66469 (#66995)
a806cc2bc43 is described below
commit a806cc2bc4381dd7bd45779b4d5a65d6a567ae47
Author: Mingyu Chen (Rayner) <[email protected]>
AuthorDate: Fri Aug 21 10:20:34 2026 +0800
branch-4.1: [fix](load)(regression-test) fix broker load label
self-conflict on pending retry; deflake P2 suites #66469 (#66995)
### What problem does this PR solve?
Issue Number: close #xxx
Related PR: #66469 (the master PR this picks), #66884 / #66920 (the
follow-up that is
**already on branch-4.1** and was written against this fix)
Problem Summary:
Cherry-pick of #66469 to branch-4.1. All six root causes it fixes also
exist on
branch-4.1 (2 FE product bugs + 3 test/CI fixes; item 5 is partly
already fixed here —
see the conflict notes).
1. **[load] Broker load pending-task retry self-conflicts on its own
label** (FE product
bug). `BrokerLoadPendingTask.executeTask()` runs `getAllFileStatus()`
then `beginTxn()`,
and `LoadTask.exec` retries the whole task on failure (retryTime=3),
with
`onTaskFinished()` inside the same try. Once the txn exists, any failure
makes every
retry hit `LabelAlreadyUsedException` against the job's OWN PREPARE txn,
burn all
retries within milliseconds and cancel the job with a misleading
`Label [...] has already been used, relate to txn [...], status
[PREPARE]` that masks
the real cause. Fix: make `beginTxn()` retry-idempotent — reuse an
already-assigned
`transactionId`, and on `LabelAlreadyUsedException` adopt the label's
txn iff it is ours
(`callbackId == job id`, status PREPARE). Foreign conflicts and lookup
failures still
rethrow the original exception. 3 Mockito unit tests.
2. **[load] Nereids planning failures must cancel the job with the real
cause** (FE product
bug, the trigger's escape route).
`org.apache.doris.nereids.exceptions.AnalysisException`
extends RuntimeException, so it bypassed `onPendingTaskFinished`'s
`catch (UserException)` (which cancels immediately with the real
message) and fell into
the generic retry path. Catch it alongside UserException. The new test
drives the real
propagation path (mocked `NereidsLoadingTaskPlanner.plan` throws through
the real
`LoadLoadingTask.init`/`createLoadingTask` chain) and asserts the job is
CANCELLED with
the real message and no loading tasks.
3. **plugin_compaction.groovy**: treat base compaction's `E-808`
(`BE_NO_SUITABLE_VERSION`)
as benign, exactly like cumulative's E-2000/E-2010. Suites still verify
the compaction
effect via their own rowset/segment-count asserts.
4. **compaction_width_array_column**: with `BUCKETS 2` the fixture loads
a ~56GB
197-segment overlapping rowset into a single tablet; compaction writes
the full output
on the same mount before deleting the input, which structurally ENOSPCs
on 100GB CI data
disks. Bump to `BUCKETS 16` (~7GB/tablet) and raise the load to 16G
exec_mem_limit /
load_parallelism 1.
5. **inverted_index_p2/test_show_data**: branch-4.1 has already fixed
three of the four
sub-suites independently (#65387, #65927). This pick adds only what is
missing here:
the three `wait_timeout` guards, and the `test_show_data_for_bkd`
comparison — the one
sub-suite branch-4.1 never fixed, which still did an exact
`assertEquals` on index sizes
produced by two different writer paths.
6.
**cold_heat_separation_p2/table_modify_resouce_and_policy(+_by_hdfs)**:
both phases
waited only for `tablets[0]` to finish cooldown, then asserted
`remote_data_size > 0`
for EVERY replica row. Replicated cooldown is
leader-uploads/followers-follow-async, so
a lagging follower failed the assert with zero grace. Make the wait
cover all replica
rows (local==0 && remote>0) with the same 100x10s budget.
### Conflicts resolved
**`BrokerLoadJob.java`** applied without conflict and is byte-identical
to master's version
of the two changed hunks. Worth calling out: branch-4.1 already carries
#66920 (pick of
#66884), which resets `transactionId = 0` in
`CloudBrokerLoadJob.unprotectedExecuteRetry()`
with the comment *"so beginTxn() does not blindly reuse a transaction
that was aborted …
beginTxn() will … adopt the transaction only after verifying that it
belongs to this job."*
That behaviour is exactly what this PR introduces, so branch-4.1
currently has the
follow-up without its base. `CloudBrokerLoadJob` does not override
`beginTxn()`, so it
inherits the fix.
**`BrokerLoadJobTest.java` (import block)** — branch-4.1 has not taken
master's
JMockit → Mockito migration (#62221), so the file is still JMockit-based
while the picked
tests are Mockito-based. Kept branch-4.1's JMockit imports and added the
imports the picked
tests need. `mockito-core` and `mockito-inline` are already test-scoped
in `fe-core` on
branch-4.1, and 18 other `fe-core` test classes already mix both
frameworks, so no
dependency or migration work is needed.
**`test_show_data.groovy`** — branch-4.1 already fixed this exact flake
in #65387 and
#65927, with a *stricter* 15% ratio gate plus an absolute size-range
gate. branch-4.1's
version was kept in all four conflicting comparisons: taking master's
text would have
loosened 15% → 20% and deleted the range assertions (leaving #65387's
helper closures
dead). Only the additive parts of the pick were taken.
### Release note
None
### Check List (For Author)
- Test <!-- At least one of them must be included. -->
- [x] Regression test
- [x] Unit Test
- [ ] Manual test (add detailed scripts or steps below)
- [ ] No need to test or manual test. Explain why:
- [ ] This is a refactor/code format and no logic has been changed.
- [ ] Previous test can cover this change.
- [ ] No code files have been changed.
- [ ] Other reason <!-- Add your reason? -->
- Behavior changed:
- [ ] No.
- [x] Yes. (a) Broker load jobs whose pending task is retried after the
job's txn was
already begun now reuse (or adopt) that txn instead of being cancelled
with a
misleading "Label has already been used" error. (b) Nereids planning
failures now
cancel the load job immediately with the real cause instead of going
through the
generic retry path. No behavior change outside those failure paths.
- Does this need documentation?
- [x] No.
- [ ] Yes. <!-- Add document PR link here. eg:
https://github.com/apache/doris-website/pull/1214 -->
---
.../apache/doris/load/loadv2/BrokerLoadJob.java | 59 ++++++-
.../doris/load/loadv2/BrokerLoadJobTest.java | 171 +++++++++++++++++++++
regression-test/plugins/plugin_compaction.groovy | 8 +-
.../table_modify_resouce_and_policy.groovy | 43 ++++--
.../table_modify_resouce_and_policy_by_hdfs.groovy | 43 ++++--
.../compaction_width_array_column.groovy | 7 +-
.../suites/compaction/ddl/column_witdh_array.sql | 5 +-
.../suites/inverted_index_p2/test_show_data.groovy | 8 +-
8 files changed, 309 insertions(+), 35 deletions(-)
diff --git
a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
index 9c7996a05cc..4fb3ac88911 100644
--- a/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
+++ b/fe/fe-core/src/main/java/org/apache/doris/load/loadv2/BrokerLoadJob.java
@@ -62,6 +62,7 @@ import org.apache.doris.transaction.BeginTransactionException;
import org.apache.doris.transaction.TransactionState;
import org.apache.doris.transaction.TransactionState.TxnCoordinator;
import org.apache.doris.transaction.TransactionState.TxnSourceType;
+import org.apache.doris.transaction.TransactionStatus;
import com.google.common.base.Joiner;
import com.google.common.collect.Lists;
@@ -129,13 +130,52 @@ public class BrokerLoadJob extends BulkLoadJob {
public void beginTxn()
throws LabelAlreadyUsedException, BeginTransactionException,
AnalysisException, DuplicatedRequestException,
QuotaExceedException, MetaNotFoundException {
- transactionId = Env.getCurrentGlobalTransactionMgr()
- .beginTransaction(dbId,
Lists.newArrayList(fileGroupAggInfo.getAllTableIds()), label, null,
- new TxnCoordinator(TxnSourceType.FE, 0,
- FrontendOptions.getLocalHostAddress(),
- ExecuteEnv.getInstance().getStartupTime()),
- TransactionState.LoadJobSourceType.BATCH_LOAD_JOB, id,
- getTimeout());
+ if (transactionId > 0) {
+ // A previous attempt of the pending task already began our txn
and failed afterwards;
+ // the retried task must reuse it instead of failing
LabelAlreadyUsedException
+ // against the job's own txn.
+ LOG.info("broker load job {} reuses already begun txn {} on
pending task retry", id, transactionId);
+ return;
+ }
+ try {
+ transactionId = Env.getCurrentGlobalTransactionMgr()
+ .beginTransaction(dbId,
Lists.newArrayList(fileGroupAggInfo.getAllTableIds()), label, null,
+ new TxnCoordinator(TxnSourceType.FE, 0,
+ FrontendOptions.getLocalHostAddress(),
+ ExecuteEnv.getInstance().getStartupTime()),
+ TransactionState.LoadJobSourceType.BATCH_LOAD_JOB,
id,
+ getTimeout());
+ } catch (LabelAlreadyUsedException e) {
+ // The label may be occupied by our OWN txn: a previous attempt
registered it but threw
+ // before transactionId was assigned (e.g. edit log write
failure), and the pending
+ // task retry would otherwise burn all retries on this exception
and cancel the job
+ // with a misleading "Label has already been used".
+ Long ownTxnId = findSelfPreparedTxnByLabel();
+ if (ownTxnId == null) {
+ throw e;
+ }
+ LOG.info("broker load job {} adopts its own prepared txn {} for
label {} on pending task retry",
+ id, ownTxnId, label);
+ transactionId = ownTxnId;
+ }
+ }
+
+ private Long findSelfPreparedTxnByLabel() {
+ try {
+ Long txnId =
Env.getCurrentGlobalTransactionMgr().getTransactionIdByLabel(dbId, label,
+ Lists.newArrayList(TransactionStatus.PREPARE));
+ if (txnId == null) {
+ return null;
+ }
+ TransactionState existingTxn =
Env.getCurrentGlobalTransactionMgr().getTransactionState(dbId, txnId);
+ if (existingTxn != null && existingTxn.getCallbackId() == id
+ && existingTxn.getTransactionStatus() ==
TransactionStatus.PREPARE) {
+ return txnId;
+ }
+ } catch (Exception lookupException) {
+ LOG.warn("broker load job {} failed to look up txn by label {}",
id, label, lookupException);
+ }
+ return null;
}
@Override
@@ -203,7 +243,10 @@ public class BrokerLoadJob extends BulkLoadJob {
try {
Database db = getDb();
createLoadingTask(db, attachment);
- } catch (UserException e) {
+ } catch (UserException |
org.apache.doris.nereids.exceptions.AnalysisException e) {
+ // Nereids analysis errors extend RuntimeException, not
UserException; without this
+ // branch a deterministic planning failure (e.g. "disk ... exceed
limit usage") falls
+ // into the generic pending-task retry path and the real cause is
never reported.
LOG.warn(new LogBuilder(LogKey.LOAD_JOB, id)
.add("database_id", dbId)
.add("error_msg", "Failed to divide job into loading
task.")
diff --git
a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
index 5ec71bcb43b..767c360c880 100644
---
a/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
+++
b/fe/fe-core/src/test/java/org/apache/doris/load/loadv2/BrokerLoadJobTest.java
@@ -17,10 +17,13 @@
package org.apache.doris.load.loadv2;
+import org.apache.doris.analysis.BrokerDesc;
import org.apache.doris.catalog.Database;
import org.apache.doris.catalog.Env;
import org.apache.doris.catalog.OlapTable;
import org.apache.doris.catalog.Table;
+import org.apache.doris.catalog.TableProperty;
+import org.apache.doris.common.LabelAlreadyUsedException;
import org.apache.doris.common.MetaNotFoundException;
import org.apache.doris.common.jmockit.Deencapsulation;
import org.apache.doris.datasource.InternalCatalog;
@@ -28,10 +31,19 @@ import org.apache.doris.load.BrokerFileGroup;
import org.apache.doris.load.BrokerFileGroupAggInfo;
import org.apache.doris.load.BrokerFileGroupAggInfo.FileGroupAggKey;
import org.apache.doris.load.EtlStatus;
+import org.apache.doris.load.FailMsg;
import org.apache.doris.metric.MetricRepo;
+import org.apache.doris.nereids.load.NereidsBrokerFileGroup;
import org.apache.doris.nereids.load.NereidsLoadingTaskPlanner;
+import org.apache.doris.persist.EditLog;
+import org.apache.doris.resource.computegroup.ComputeGroup;
+import org.apache.doris.resource.computegroup.ComputeGroupMgr;
import org.apache.doris.task.MasterTaskExecutor;
+import org.apache.doris.thrift.TBrokerFileStatus;
+import org.apache.doris.transaction.GlobalTransactionMgrIface;
import org.apache.doris.transaction.TransactionState;
+import org.apache.doris.transaction.TransactionStatus;
+import org.apache.doris.transaction.TxnStateCallbackFactory;
import com.google.common.collect.Lists;
import com.google.common.collect.Maps;
@@ -42,7 +54,11 @@ import mockit.Mocked;
import org.junit.Assert;
import org.junit.BeforeClass;
import org.junit.Test;
+import org.mockito.MockedConstruction;
+import org.mockito.MockedStatic;
+import org.mockito.Mockito;
+import java.util.Collections;
import java.util.List;
import java.util.Map;
import java.util.Optional;
@@ -375,4 +391,159 @@ public class BrokerLoadJobTest {
Assert.assertEquals(1, brokerLoadJob.getFinishTimestamp());
Assert.assertEquals(JobState.LOADING, brokerLoadJob.getState());
}
+
+ @Test
+ public void testBeginTxnReusesAlreadyBegunTxn() throws Exception {
+ // A retried pending task must not begin a second txn for the same job
(it would fail
+ // LabelAlreadyUsedException against the job's own txn).
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "transactionId", 12345L);
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ brokerLoadJob.beginTxn();
+ }
+
+ Assert.assertEquals(12345L, (long)
Deencapsulation.getField(brokerLoadJob, "transactionId"));
+ Mockito.verifyNoInteractions(transactionMgr);
+ }
+
+ @Test
+ public void testBeginTxnAdoptsOwnPreparedTxn() throws Exception {
+ // First attempt registered the txn but threw before transactionId was
assigned
+ // (e.g. edit log write failure); the retry gets
LabelAlreadyUsedException for the job's
+ // OWN prepared txn and must adopt it instead of cancelling the job.
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TransactionState preparedTxn = Mockito.mock(TransactionState.class);
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "id", 1001L);
+ Deencapsulation.setField(brokerLoadJob, "dbId", 1L);
+ Deencapsulation.setField(brokerLoadJob, "label",
"label_self_conflict");
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ Mockito.when(transactionMgr.beginTransaction(Mockito.anyLong(),
Mockito.anyList(),
+ Mockito.anyString(), Mockito.any(), Mockito.any(),
Mockito.any(),
+ Mockito.anyLong(), Mockito.anyLong()))
+ .thenThrow(new
LabelAlreadyUsedException("label_self_conflict"));
+
Mockito.when(transactionMgr.getTransactionIdByLabel(Mockito.anyLong(),
Mockito.anyString(),
+ Mockito.anyList())).thenReturn(777L);
+ Mockito.when(transactionMgr.getTransactionState(Mockito.anyLong(),
Mockito.eq(777L)))
+ .thenReturn(preparedTxn);
+ Mockito.when(preparedTxn.getCallbackId()).thenReturn(1001L);
+
Mockito.when(preparedTxn.getTransactionStatus()).thenReturn(TransactionStatus.PREPARE);
+
+ brokerLoadJob.beginTxn();
+ }
+
+ Assert.assertEquals(777L, (long)
Deencapsulation.getField(brokerLoadJob, "transactionId"));
+ }
+
+ @Test
+ public void testBeginTxnRethrowsForeignLabelConflict() throws Exception {
+ // The label belongs to some other job's txn: the original exception
must propagate.
+ GlobalTransactionMgrIface transactionMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ TransactionState foreignTxn = Mockito.mock(TransactionState.class);
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "id", 1001L);
+ Deencapsulation.setField(brokerLoadJob, "dbId", 1L);
+ Deencapsulation.setField(brokerLoadJob, "label", "label_foreign");
+
+ try (MockedStatic<Env> envMockedStatic =
Mockito.mockStatic(Env.class)) {
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(transactionMgr);
+ Mockito.when(transactionMgr.beginTransaction(Mockito.anyLong(),
Mockito.anyList(),
+ Mockito.anyString(), Mockito.any(), Mockito.any(),
Mockito.any(),
+ Mockito.anyLong(), Mockito.anyLong()))
+ .thenThrow(new LabelAlreadyUsedException("label_foreign"));
+
Mockito.when(transactionMgr.getTransactionIdByLabel(Mockito.anyLong(),
Mockito.anyString(),
+ Mockito.anyList())).thenReturn(888L);
+ Mockito.when(transactionMgr.getTransactionState(Mockito.anyLong(),
Mockito.eq(888L)))
+ .thenReturn(foreignTxn);
+ Mockito.when(foreignTxn.getCallbackId()).thenReturn(9999L);
+
+ try {
+ brokerLoadJob.beginTxn();
+ Assert.fail("expected LabelAlreadyUsedException");
+ } catch (LabelAlreadyUsedException expected) {
+ // expected
+ }
+ }
+
+ Assert.assertEquals(0L, (long) Deencapsulation.getField(brokerLoadJob,
"transactionId"));
+ }
+
+ @Test
+ public void testPendingTaskOnFinishedWithNereidsPlanningError() throws
Exception {
+ // A Nereids planning error is a RuntimeException; it must cancel the
job with the real
+ // cause instead of escaping into the generic pending-task retry path
(which used to end
+ // in a misleading "Label has already been used" cancellation).
+ BrokerPendingTaskAttachment attachment =
Mockito.mock(BrokerPendingTaskAttachment.class);
+ Env env = Mockito.mock(Env.class);
+ InternalCatalog catalog = Mockito.mock(InternalCatalog.class);
+ Database database = Mockito.mock(Database.class);
+ BrokerFileGroupAggInfo fileGroupAggInfo =
Mockito.mock(BrokerFileGroupAggInfo.class);
+ BrokerFileGroup brokerFileGroup = Mockito.mock(BrokerFileGroup.class);
+ NereidsBrokerFileGroup nereidsBfg =
Mockito.mock(NereidsBrokerFileGroup.class);
+
Mockito.when(brokerFileGroup.toNereidsBrokerFileGroup()).thenReturn(nereidsBfg);
+ OlapTable olapTable = Mockito.mock(OlapTable.class);
+ GlobalTransactionMgrIface globalTxnMgr =
Mockito.mock(GlobalTransactionMgrIface.class);
+ ProgressManager progressManager = Mockito.mock(ProgressManager.class);
+ ComputeGroupMgr computeGroupMgr = Mockito.mock(ComputeGroupMgr.class);
+ TableProperty tableProperty = Mockito.mock(TableProperty.class);
+
+ try (MockedStatic<Env> envMockedStatic = Mockito.mockStatic(Env.class);
+ MockedConstruction<NereidsLoadingTaskPlanner> ignored =
+
Mockito.mockConstruction(NereidsLoadingTaskPlanner.class, (mock, context) ->
+ Mockito.doThrow(new
org.apache.doris.nereids.exceptions.AnalysisException(
+ "disk /mnt/mock on backend 10001
exceed limit usage"))
+ .when(mock).plan(Mockito.any(),
Mockito.anyList(), Mockito.anyInt()))) {
+ envMockedStatic.when(Env::getCurrentEnv).thenReturn(env);
+
envMockedStatic.when(Env::getCurrentInternalCatalog).thenReturn(catalog);
+
envMockedStatic.when(Env::getCurrentProgressManager).thenReturn(progressManager);
+
envMockedStatic.when(Env::getCurrentGlobalTransactionMgr).thenReturn(globalTxnMgr);
+
+ BrokerLoadJob brokerLoadJob = new BrokerLoadJob();
+ Deencapsulation.setField(brokerLoadJob, "state", JobState.LOADING);
+ BrokerDesc brokerDesc = Mockito.mock(BrokerDesc.class);
+ Deencapsulation.setField(brokerLoadJob, "brokerDesc", brokerDesc);
+
+ Map<FileGroupAggKey, List<BrokerFileGroup>> aggKeyToFileGroups =
Maps.newHashMap();
+ FileGroupAggKey aggKey = new FileGroupAggKey(1L, null);
+ aggKeyToFileGroups.put(aggKey,
Lists.newArrayList(brokerFileGroup));
+ Deencapsulation.setField(brokerLoadJob, "fileGroupAggInfo",
fileGroupAggInfo);
+
+ Mockito.when(attachment.getTaskId()).thenReturn(1L);
+
Mockito.doReturn(database).when(catalog).getDbOrMetaException(Mockito.anyLong());
+ Mockito.doReturn(Lists.newArrayList()).when(database)
+ .getTablesOnIdOrderOrThrowException(Mockito.anyList());
+
Mockito.when(fileGroupAggInfo.getAggKeyToFileGroups()).thenReturn(aggKeyToFileGroups);
+
Mockito.when(fileGroupAggInfo.getAllTableIds()).thenReturn(Sets.newHashSet(1L));
+
Mockito.doReturn(olapTable).when(database).getTableNullable(Mockito.anyLong());
+ Mockito.when(olapTable.isTemporary()).thenReturn(false);
+
Mockito.when(olapTable.getTableProperty()).thenReturn(tableProperty);
+
Mockito.when(tableProperty.getUseSchemaLightChange()).thenReturn(false);
+ Mockito.when(olapTable.getIndexes()).thenReturn(null);
+ Mockito.when(attachment.getFileStatusByTable(aggKey)).thenReturn(
+ Collections.singletonList(Collections.singletonList(new
TBrokerFileStatus())));
+ Mockito.when(attachment.getFileNumByTable(aggKey)).thenReturn(1);
+ Mockito.when(env.getNextId()).thenReturn(1L);
+ Mockito.when(env.getComputeGroupMgr()).thenReturn(computeGroupMgr);
+ Mockito.when(env.getInternalCatalog()).thenReturn(catalog);
+ EditLog editLog = Mockito.mock(EditLog.class);
+ Mockito.when(env.getEditLog()).thenReturn(editLog);
+ Mockito.when(computeGroupMgr.getAllBackendComputeGroup())
+ .thenReturn(new ComputeGroup("default", "default", null));
+ TxnStateCallbackFactory callbackFactory =
Mockito.mock(TxnStateCallbackFactory.class);
+
Mockito.when(globalTxnMgr.getCallbackFactory()).thenReturn(callbackFactory);
+
+ brokerLoadJob.onTaskFinished(attachment);
+
+ Assert.assertEquals(JobState.CANCELLED, brokerLoadJob.getState());
+ FailMsg failMsg = Deencapsulation.getField(brokerLoadJob,
"failMsg");
+ Assert.assertTrue(failMsg.getMsg().contains("exceed limit usage"));
+ Map<Long, LoadTask> idToTasks =
Deencapsulation.getField(brokerLoadJob, "idToTasks");
+ Assert.assertEquals(0, idToTasks.size());
+ }
+ }
}
diff --git a/regression-test/plugins/plugin_compaction.groovy
b/regression-test/plugins/plugin_compaction.groovy
index 42602699149..8cf6b1f0f87 100644
--- a/regression-test/plugins/plugin_compaction.groovy
+++ b/regression-test/plugins/plugin_compaction.groovy
@@ -106,8 +106,14 @@ Suite.metaClass.trigger_and_wait_compaction = { String
table_name, String compac
triggered_tablets.add(tablet) // compaction already in queue,
treat it as successfully triggered
} else if (!auto_compaction_disabled) {
// ignore the error if auto compaction enabled
- } else if (status_lower.contains("e-2000") ||
status_lower.contains("e-2010")) {
+ } else if (status_lower.contains("e-2000") ||
status_lower.contains("e-2010")
+ || status_lower.contains("e-808")) {
// ignore this tablet compaction.
+ // e-2000/e-2010: cumulative has no suitable version;
+ // e-808 (BE_NO_SUITABLE_VERSION): base compaction has nothing
to merge on this
+ // replica (e.g. only [0-1]+[2-y] with an empty [0-1]) — a
by-design no-op, the
+ // base analogue of e-2000. Replica layouts can legitimately
diverge here when a
+ // lagging publish made an earlier cumulative trigger a no-op
on one replica.
} else if (ignored_errors.any { error ->
status_lower.contains(error.toLowerCase()) }) {
// ignore this tablet compaction if the error is in the
ignored_errors list
} else {
diff --git
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
index 94600dbb281..94e309c2b30 100644
---
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
+++
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy.groovy
@@ -197,18 +197,28 @@ suite("table_modify_resouce") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
-
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
def try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
@@ -281,17 +291,28 @@ suite("table_modify_resouce") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
diff --git
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
index 0feda8f0845..89a06aad359 100644
---
a/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
+++
b/regression-test/suites/cold_heat_separation_p2/table_modify_resouce_and_policy_by_hdfs.groovy
@@ -193,18 +193,28 @@ suite("table_modify_resouce_by_hdfs") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
-
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
def try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
@@ -277,17 +287,28 @@ suite("table_modify_resouce_by_hdfs") {
"""
log.info( "test tablets not empty")
assertTrue(tablets.size() > 0)
- fetchDataSize(sizes, tablets[0])
+ // Cooldown on a replicated table is leader-uploads /
followers-follow-async: the asserts
+ // below check EVERY replica row, so the wait must too, not just
tablets[0].
try_times = 100
- while (sizes[0] != 0) {
- log.info( "test local size is not zero, sleep 10s")
+ while (true) {
+ def all_cooled = true
+ for (def tablet in tablets) {
+ fetchDataSize(sizes, tablet)
+ if (sizes[0] != 0 || sizes[1] <= 0) {
+ all_cooled = false
+ break
+ }
+ }
+ if (all_cooled) {
+ break
+ }
+ log.info( "not all replicas cooled down, sleep 10s")
sleep(10000)
tablets = sql_return_maparray """
SHOW TABLETS FROM ${tableName}
"""
- fetchDataSize(sizes, tablets[0])
try_times -= 1
- assertTrue(try_times > 0, "remote size is still zero, maybe some error
occurred")
+ assertTrue(try_times > 0, "cooldown not finished on all replicas,
maybe some error occurred")
}
// 修改resource和policy到新值然后查看remote data size是否能对上
diff --git
a/regression-test/suites/compaction/compaction_width_array_column.groovy
b/regression-test/suites/compaction/compaction_width_array_column.groovy
index d80e67c6cd8..225c412896b 100644
--- a/regression-test/suites/compaction/compaction_width_array_column.groovy
+++ b/regression-test/suites/compaction/compaction_width_array_column.groovy
@@ -33,14 +33,17 @@ suite('compaction_width_array_column', "p2") {
def s3BucketName = getS3BucketName()
def random = new Random();
+ // Loading wide array columns is memory-intensive: the default 8G
exec_mem_limit was exceeded
+ // (peak ~12G) and the load got cancelled with MEM_LIMIT_EXCEEDED. Raise
the limit to 16G and
+ // load with parallelism 1 to keep the per-load peak comfortably under the
limit.
def s3WithProperties = """WITH S3 (
|"AWS_ACCESS_KEY" = "${getS3AK()}",
|"AWS_SECRET_KEY" = "${getS3SK()}",
|"AWS_ENDPOINT" = "${getS3Endpoint()}",
|"AWS_REGION" = "${getS3Region()}")
|PROPERTIES(
- |"exec_mem_limit" = "8589934592",
- |"load_parallelism" = "3")""".stripMargin()
+ |"exec_mem_limit" = "17179869184",
+ |"load_parallelism" = "1")""".stripMargin()
// set fe configuration
sql "ADMIN SET FRONTEND CONFIG ('max_bytes_per_broker_scanner' =
'161061273600')"
diff --git a/regression-test/suites/compaction/ddl/column_witdh_array.sql
b/regression-test/suites/compaction/ddl/column_witdh_array.sql
index b51345a2084..c857221cdde 100644
--- a/regression-test/suites/compaction/ddl/column_witdh_array.sql
+++ b/regression-test/suites/compaction/ddl/column_witdh_array.sql
@@ -5,7 +5,10 @@ CREATE TABLE column_witdh_array (
k2 JSON NULL
) ENGINE=OLAP
DUPLICATE KEY(street, streetaddress)
-DISTRIBUTED BY HASH(street) BUCKETS 2
+-- 16 buckets keeps each tablet's single loaded rowset around 7GB; with 2
buckets it was ~56GB,
+-- and compaction (which writes the full output on the same mount before
deleting the input)
+-- structurally ENOSPCed on 100GB CI data disks.
+DISTRIBUTED BY HASH(street) BUCKETS 16
PROPERTIES (
"replication_num" = "1",
"disable_auto_compaction" = "true"
diff --git a/regression-test/suites/inverted_index_p2/test_show_data.groovy
b/regression-test/suites/inverted_index_p2/test_show_data.groovy
index 6121f288569..8f987858046 100644
--- a/regression-test/suites/inverted_index_p2/test_show_data.groovy
+++ b/regression-test/suites/inverted_index_p2/test_show_data.groovy
@@ -204,6 +204,7 @@ suite("test_show_data", "p2") {
create_httplogs_table_with_index.call(testTableWithIndex)
load_httplogs_data.call(testTableWithIndex,
'test_httplogs_load_with_index', 'true', 'json', 'documents-1000.json')
def another_with_index_size =
wait_for_show_data_finish(testTableWithIndex, 60000, 0)
+ assertTrue(another_with_index_size != "wait_timeout")
if (!isCloudMode()) {
double diff = Math.abs(another_with_index_size.toDouble() -
with_index_size.toDouble())
double max_size = Math.max(another_with_index_size.toDouble(),
with_index_size.toDouble())
@@ -410,8 +411,12 @@ suite("test_show_data_for_bkd", "p2") {
create_httplogs_table_with_bkd_index.call(testTableWithBKDIndex)
load_httplogs_data.call(testTableWithBKDIndex,
'test_httplogs_load_with_bkd_index', 'true', 'json', 'documents-1000.json')
def another_with_index_size =
wait_for_show_data_finish(testTableWithBKDIndex, 60000, 0)
+ assertTrue(another_with_index_size != "wait_timeout")
if (!isCloudMode()) {
- assertEquals(another_with_index_size, with_index_size)
+ // Same rationale as test_show_data: writer-path dependent index
size, 20% tolerance.
+ assertTrue(Math.abs(another_with_index_size - with_index_size)
+ <= 0.2 * Math.max(another_with_index_size,
with_index_size),
+ "index size mismatch beyond 20% tolerance:
inline_index=${another_with_index_size}, built_index=${with_index_size}")
}
} finally {
//try_sql("DROP TABLE IF EXISTS ${testTable}")
@@ -613,6 +618,7 @@ suite("test_show_data_multi_add", "p2") {
create_httplogs_table_with_index.call(testTableWithIndex)
load_httplogs_data.call(testTableWithIndex,
'test_show_data_httplogs_multi_add_with_index', 'true', 'json',
'documents-1000.json')
def another_with_index_size =
wait_for_show_data_finish(testTableWithIndex, 60000, 0)
+ assertTrue(another_with_index_size != "wait_timeout")
if (!isCloudMode()) {
double diff = Math.abs(another_with_index_size.toDouble() -
with_index_size2.toDouble())
double max_size = Math.max(another_with_index_size.toDouble(),
with_index_size2.toDouble())
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
