[BugFix] Fix schema change publish does not retry in shared-data (backport #64093) (#64204)

Signed-off-by: PengFei Li <[email protected]>
Co-authored-by: PengFei Li <[email protected]>

Author: mergify[bot]

fe/fe-core/src/main/java/com/starrocks/alter/AlterJobV2.java

@@ -348,13 +348,16 @@ protected boolean publishVersion() {
                 return lakePublishVersion();
             };
             publishVersionFuture = GlobalStateMgr.getCurrentState().getLakeAlterPublishExecutor().submit(task);
+            LOG.info("submit publish task for job: {}", jobId);
             return false;
         } else {
             if (publishVersionFuture.isDone()) {
                 try {
                     return publishVersionFuture.get();
                 } catch (InterruptedException | ExecutionException e) {
                     return false;
+                } finally {
+                    publishVersionFuture = null;
                 }
             } else {
                 return false;

fe/fe-core/src/test/java/com/starrocks/alter/LakeTableSchemaChangeJobTest.java

@@ -46,9 +46,12 @@
 import org.junit.jupiter.api.Test;
 
 import java.io.IOException;
+import java.io.PrintWriter;
+import java.io.StringWriter;
 import java.util.Collection;
 import java.util.List;
 import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.concurrent.atomic.AtomicInteger;
 
 public class LakeTableSchemaChangeJobTest {
     private static final int NUM_BUCKETS = 4;
@@ -65,6 +68,16 @@ public LakeTableSchemaChangeJobTest() {
     public static void setUp() throws Exception {
         UtFrameUtils.createMinStarRocksCluster(RunMode.SHARED_DATA);
         connectContext = UtFrameUtils.createDefaultCtx();
+        // Schema change job is executed manually, so stop the schema change daemon to avoid interfering with the test.
+        SchemaChangeHandler schemaChangeHandler = GlobalStateMgr.getCurrentState().getAlterJobMgr().getSchemaChangeHandler();
+        schemaChangeHandler.setStop();
+        schemaChangeHandler.interrupt();
+        long startTime = System.currentTimeMillis();
+        while (schemaChangeHandler.isRunning()) {
+            Thread.sleep(100);
+            Assertions.assertTrue(System.currentTimeMillis() - startTime < 60000,
+                    "Schema change handler is not stopped in 60 seconds");
+        }
     }
 
     private static LakeTable createTable(ConnectContext connectContext, String sql) throws Exception {
@@ -489,6 +502,56 @@ public boolean publishVersion() {
         Assertions.assertEquals(AlterJobV2.JobState.FINISHED, schemaChangeJob.getJobState());
     }
 
+    @Test
+    public void testPublishRetry() throws Exception {
+        AtomicInteger numPublishRetry = new AtomicInteger(0);
+        new MockUp<LakeTableSchemaChangeJob>() {
+            @Mock
+            public boolean lakePublishVersion() {
+                numPublishRetry.incrementAndGet();
+                return false;
+            }
+        };
+        
+        new MockUp<LakeTableSchemaChangeJob>() {
+            @Mock
+            public void sendAgentTask(AgentBatchTask batchTask) {
+                batchTask.getAllTasks().forEach(t -> t.setFinished(true));
+            }
+        };
+
+        schemaChangeJob.run();
+        Assertions.assertEquals(AlterJobV2.JobState.FINISHED_REWRITING, schemaChangeJob.getJobState());
+        long timeoutMs = 10 * 60 * 1000L;
+        long deadline = System.currentTimeMillis() + timeoutMs;
+        while (numPublishRetry.get() < 2) {
+            schemaChangeJob.run();
+            Thread.sleep(100);
+            Assertions.assertTrue(System.currentTimeMillis() < deadline,
+                    () -> String.format("publish does not retry before timeout, job state: %s", schemaChangeJob.getJobState()));
+        }
+
+        new MockUp<LakeTableSchemaChangeJob>() {
+            @Mock
+            public boolean lakePublishVersion() {
+                return true;
+            }
+        };
+
+        deadline = System.currentTimeMillis() + timeoutMs;
+        while (schemaChangeJob.getJobState() != AlterJobV2.JobState.FINISHED
+                || table.getState() != OlapTable.OlapTableState.NORMAL) {
+            schemaChangeJob.run();
+            Thread.sleep(100);
+            Assertions.assertTrue(System.currentTimeMillis() < deadline,
+                    () -> String.format("publish does not finish before timeout, job state: %s, table state: %s",
+                            schemaChangeJob.getJobState(), table.getState()));
+        }
+        Assertions.assertEquals(2, table.getBaseSchema().size());
+        Assertions.assertEquals("c0", table.getBaseSchema().get(0).getName());
+        Assertions.assertEquals("c1", table.getBaseSchema().get(1).getName());
+    }
+
     @Test
     public void testTransactionRaceCondition() throws AlterCancelException {
         new MockUp<LakeTableSchemaChangeJob>() {
@@ -516,7 +579,13 @@ public boolean isPreviousLoadFinished(long dbId, long tableId, long txnId) throw
         Exception exception = Assertions.assertThrows(AlterCancelException.class, () ->
                 schemaChangeJob.runPendingJob());
         Assertions.assertTrue(exception.getMessage().contains(
-                    "concurrent transaction detected while adding shadow index, please re-run the alter table command"));
+                "concurrent transaction detected while adding shadow index, please re-run the alter table command"),
+                () -> {
+                    StringWriter sw = new java.io.StringWriter();
+                    PrintWriter pw = new java.io.PrintWriter(sw);
+                    exception.printStackTrace(pw);
+                    return sw.toString();
+                });
         Assertions.assertEquals(AlterJobV2.JobState.PENDING, schemaChangeJob.getJobState());
         Assertions.assertEquals(10101L, schemaChangeJob.getWatershedTxnId());