[BugFix] Fix replaying AlterJob NPE Bug (backport #50100) (#50107)

Signed-off-by: gengjun-git <[email protected]> Co-authored-by: gengjun-git <[email protected]>
StarRocks · Aug 21, 2024 · f61f51a · f61f51a
1 parent e5fc13b
commit f61f51a
Show file tree

Hide file tree

Showing 6 changed files with 27 additions and 4 deletions.
diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAlterMetaJob.java b/fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAlterMetaJob.java
@@ -37,6 +37,11 @@ public class LakeTableAlterMetaJob extends LakeTableAlterMetaJobBase {
     @SerializedName(value = "metaValue")
     private boolean metaValue;
 
+    // for deserialization
+    public LakeTableAlterMetaJob() {
+        super(JobType.SCHEMA_CHANGE);
+    }
+
     public LakeTableAlterMetaJob(long jobId, long dbId, long tableId, String tableName,
                                  long timeoutMs, TTabletMetaType metaType, boolean metaValue) {
         super(jobId, JobType.SCHEMA_CHANGE, dbId, tableId, tableName, timeoutMs);

diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAlterMetaJobBase.java b/fe/fe-core/src/main/java/com/starrocks/alter/LakeTableAlterMetaJobBase.java
@@ -68,6 +68,10 @@ public abstract class LakeTableAlterMetaJobBase extends AlterJobV2 {
     private Map<Long, Long> commitVersionMap = new HashMap<>();
     private AgentBatchTask batchTask = null;
 
+    public LakeTableAlterMetaJobBase(JobType jobType) {
+        super(jobType);
+    }
+
     public LakeTableAlterMetaJobBase(long jobId, JobType jobType, long dbId, long tableId,
                                                   String tableName, long timeoutMs) {
         super(jobId, jobType, dbId, tableId, tableName, timeoutMs);

diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/LakeTableSchemaChangeJob.java b/fe/fe-core/src/main/java/com/starrocks/alter/LakeTableSchemaChangeJob.java
@@ -138,6 +138,11 @@ public class LakeTableSchemaChangeJob extends AlterJobV2 {
     // save all schema change tasks
     private AgentBatchTask schemaChangeBatchTask = new AgentBatchTask();
 
+    // for deserialization
+    public LakeTableSchemaChangeJob() {
+        super(JobType.SCHEMA_CHANGE);
+    }
+
     public LakeTableSchemaChangeJob(long jobId, long dbId, long tableId, String tableName, long timeoutMs) {
         super(jobId, JobType.SCHEMA_CHANGE, dbId, tableId, tableName, timeoutMs);
     }

diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/OptimizeJobV2.java b/fe/fe-core/src/main/java/com/starrocks/alter/OptimizeJobV2.java
@@ -69,7 +69,7 @@ public class OptimizeJobV2 extends AlterJobV2 implements GsonPostProcessable {
     @SerializedName(value = "watershedTxnId")
     protected long watershedTxnId = -1;
 
-    private final String postfix;
+    private String postfix;
 
     @SerializedName(value = "tmpPartitionIds")
     private List<Long> tmpPartitionIds = Lists.newArrayList();
@@ -98,6 +98,11 @@ public class OptimizeJobV2 extends AlterJobV2 implements GsonPostProcessable {
     @SerializedName(value = "optimizeOperation")
     private String optimizeOperation = "";
 
+    // for deserialization
+    public OptimizeJobV2() {
+        super(JobType.OPTIMIZE);
+    }
+
     public OptimizeJobV2(long jobId, long dbId, long tableId, String tableName, long timeoutMs,
                          OptimizeClause optimizeClause) {
         this(jobId, dbId, tableId, tableName, timeoutMs);
@@ -774,9 +779,7 @@ public void write(DataOutput out) throws IOException {
 
     @Override
     public void gsonPostProcess() throws IOException {
-        if (jobState != JobState.PENDING) {
-            return;
-        }
+        this.postfix = "_" + jobId;
     }
 
     @Override

diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/RollupJobV2.java b/fe/fe-core/src/main/java/com/starrocks/alter/RollupJobV2.java
@@ -164,6 +164,11 @@ public class RollupJobV2 extends AlterJobV2 implements GsonPostProcessable {
     // save all create rollup tasks
     private AgentBatchTask rollupBatchTask = new AgentBatchTask();
 
+    // for deserialization
+    public RollupJobV2() {
+        super(JobType.ROLLUP);
+    }
+
     public RollupJobV2(long jobId, long dbId, long tableId, String tableName, long timeoutMs,
                        long baseIndexId, long rollupIndexId, String baseIndexName, String rollupIndexName,
                        int rollupSchemaVersion, List<Column> rollupSchema, Expr whereClause, int baseSchemaHash, 

diff --git a/fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeJobV2.java b/fe/fe-core/src/main/java/com/starrocks/alter/SchemaChangeJobV2.java
@@ -183,6 +183,7 @@ public SchemaChangeJobV2(long jobId, long dbId, long tableId, String tableName,
         super(jobId, JobType.SCHEMA_CHANGE, dbId, tableId, tableName, timeoutMs);
     }
 
+    // for deserialization
     private SchemaChangeJobV2() {
         super(JobType.SCHEMA_CHANGE);
     }