database/doris
2
0
Fork 0
Code Issues Pull Requests Actions 3 Packages Projects Releases Wiki Activity

[Bug] Fix bug that replayCreateLoadJob will cause fe memory leak in non master node because InsertLoadJob cannot be removed from TxnStateCallbackFactory (#6795)

Browse Source
This commit is contained in:
caiconghui
2021-10-08 13:17:22 +08:00
committed by GitHub
parent 53fed2d35e
commit 9e67b3a392
5 changed files with 13 additions and 9 deletions
Download Patch File Download Diff File Expand all files Collapse all files

3
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/InsertLoadJob.java
View File

@ -48,10 +48,11 @@ public class InsertLoadJob extends LoadJob {
super(EtlJobType.INSERT);
}
public InsertLoadJob(String label, long dbId, long tableId, long createTimestamp, String failMsg,
public InsertLoadJob(String label, long transactionId, long dbId, long tableId, long createTimestamp, String failMsg,
String trackingUrl) throws MetaNotFoundException {
super(EtlJobType.INSERT, dbId, label);
this.tableId = tableId;
this.transactionId = transactionId;
this.createTimestamp = createTimestamp;
this.loadStartTimestamp = createTimestamp;
this.finishTimestamp = System.currentTimeMillis();

10
fe/fe-core/src/main/java/org/apache/doris/load/loadv2/LoadManager.java
View File

@ -260,9 +260,11 @@ public class LoadManager implements Writable{
return;
}
addLoadJob(loadJob);
// add callback before txn created, because callback will be performed on replay without txn begin
// add callback before txn if load job is uncompleted, because callback will be performed on replay without txn begin
// register txn state listener
Catalog.getCurrentGlobalTransactionMgr().getCallbackFactory().addCallback(loadJob);
if (!loadJob.isCompleted()) {
Catalog.getCurrentGlobalTransactionMgr().getCallbackFactory().addCallback(loadJob);
}
}
private void addLoadJob(LoadJob loadJob) {
@ -278,7 +280,7 @@ public class LoadManager implements Writable{
labelToLoadJobs.get(loadJob.getLabel()).add(loadJob);
}
public void recordFinishedLoadJob(String label, String dbName, long tableId, EtlJobType jobType,
public void recordFinishedLoadJob(String label, long transactionId, String dbName, long tableId, EtlJobType jobType,
long createTimestamp, String failMsg, String trackingUrl) throws MetaNotFoundException {
// get db id
@ -287,7 +289,7 @@ public class LoadManager implements Writable{
LoadJob loadJob;
switch (jobType) {
case INSERT:
loadJob = new InsertLoadJob(label, db.getId(), tableId, createTimestamp, failMsg, trackingUrl);
loadJob = new InsertLoadJob(label, transactionId, db.getId(), tableId, createTimestamp, failMsg, trackingUrl);
break;
default:
return;

3
fe/fe-core/src/main/java/org/apache/doris/qe/StmtExecutor.java
View File

@ -1288,9 +1288,11 @@ public class StmtExecutor implements ProfileWriter {
message = throwable.getMessage();
}
txnId = insertStmt.getTransactionId();
try {
context.getCatalog().getLoadManager().recordFinishedLoadJob(
label,
txnId,
insertStmt.getDb(),
insertStmt.getTargetTable().getId(),
EtlJobType.INSERT,
@ -1301,7 +1303,6 @@ public class StmtExecutor implements ProfileWriter {
LOG.warn("Record info of insert load with error {}", e.getMessage(), e);
errMsg = "Record info of insert load with error " + e.getMessage();
}
txnId = insertStmt.getTransactionId();
}
// {'label':'my_label1', 'status':'visible', 'txnId':'123'}

2
fe/fe-core/src/test/java/org/apache/doris/load/loadv2/InsertLoadJobTest.java
View File

@ -39,7 +39,7 @@ public class InsertLoadJobTest {
public void testGetTableNames(@Mocked Catalog catalog,
@Injectable Database database,
@Injectable Table table) throws MetaNotFoundException {
InsertLoadJob insertLoadJob = new InsertLoadJob("label", 1L, 1L, 1000, "", "");
InsertLoadJob insertLoadJob = new InsertLoadJob("label", 1L, 1L, 1L, 1000, "", "");
String tableName = "table1";
new Expectations() {
{

4
fe/fe-core/src/test/java/org/apache/doris/load/loadv2/LoadManagerTest.java
View File

@ -133,7 +133,7 @@ public class LoadManagerTest {
};
loadManager = new LoadManager(new LoadJobScheduler());
LoadJob job1 = new InsertLoadJob("job1", 1L, 1L, System.currentTimeMillis(), "", "");
LoadJob job1 = new InsertLoadJob("job1", 1L, 1L, 1L, System.currentTimeMillis(), "", "");
Deencapsulation.invoke(loadManager, "addLoadJob", job1);
File file = serializeToFile(loadManager);
@ -168,7 +168,7 @@ public class LoadManagerTest {
};
loadManager = new LoadManager(new LoadJobScheduler());
LoadJob job1 = new InsertLoadJob("job1", 1L, 1L, System.currentTimeMillis(), "", "");
LoadJob job1 = new InsertLoadJob("job1", 1L, 1L, 1L, System.currentTimeMillis(), "", "");
Deencapsulation.invoke(loadManager, "addLoadJob", job1);
//make job1 don't serialize