boolean publishVersion() {
        try {
            TxnInfoPB txnInfo = new TxnInfoPB();
            txnInfo.txnId = watershedTxnId;
            txnInfo.combinedTxnLog = false;
            txnInfo.txnType = TxnTypePB.TXN_NORMAL;
            txnInfo.commitTime = finishedTimeMs / 1000;
            for (long partitionId : physicalPartitionIndexMap.rowKeySet()) {
                long commitVersion = commitVersionMap.get(partitionId);
                Map<Long, MaterializedIndex> shadowIndexMap = physicalPartitionIndexMap.row(partitionId);
                for (MaterializedIndex shadowIndex : shadowIndexMap.values()) {
                    Utils.publishVersion(shadowIndex.getTablets(), txnInfo, 1, commitVersion, warehouseId);
                }
            }
            return true;
        } catch (Exception e) {
            LOG.error("Fail to publish version for schema change job {}: {}", jobId, e.getMessage());
            return false;
        }
    }
@Test
    public void testPublishVersion() throws AlterCancelException {
        new MockUp<Utils>() {
            @Mock
            public void publishVersion(@NotNull List<Tablet> tablets, TxnInfoPB txnInfo, long baseVersion,
                                       long newVersion, long warehouseId)
                    throws
                    RpcException {
                throw new RpcException("publish version failed", "127.0.0.1");
            }
        };

        new MockUp<LakeTableSchemaChangeJob>() {
            @Mock
            public void sendAgentTask(AgentBatchTask batchTask) {
                batchTask.getAllTasks().forEach(t -> t.setFinished(true));
            }
        };

        schemaChangeJob.runPendingJob();
        Assert.assertEquals(AlterJobV2.JobState.WAITING_TXN, schemaChangeJob.getJobState());

        schemaChangeJob.runWaitingTxnJob();
        Assert.assertEquals(AlterJobV2.JobState.RUNNING, schemaChangeJob.getJobState());

        Collection<Partition> partitions = table.getPartitions();
        Assert.assertEquals(1, partitions.size());
        Partition partition = partitions.stream().findFirst().orElse(null);
        Assert.assertNotNull(partition);

        Assert.assertEquals(1, partition.getVisibleVersion());
        Assert.assertEquals(2, partition.getNextVersion());
        // Disable send publish version
        partition.setNextVersion(3);

        schemaChangeJob.runRunningJob();
        Assert.assertEquals(AlterJobV2.JobState.FINISHED_REWRITING, schemaChangeJob.getJobState());

        List<MaterializedIndex> shadowIndexes =
                partition.getMaterializedIndices(MaterializedIndex.IndexExtState.SHADOW);
        Assert.assertEquals(1, shadowIndexes.size());

        // The partition's visible version has not catch up with the commit version of this schema change job now.
        schemaChangeJob.runFinishedRewritingJob();
        Assert.assertEquals(AlterJobV2.JobState.FINISHED_REWRITING, schemaChangeJob.getJobState());

        // Reset partition's next version
        partition.setVisibleVersion(2, System.currentTimeMillis());

        // Drop table
        db.dropTable(table.getName());

        Exception exception = Assert.assertThrows(AlterCancelException.class, () -> {
            schemaChangeJob.runFinishedRewritingJob();
        });
        Assert.assertTrue(exception.getMessage().contains("Table does not exist"));
        Assert.assertEquals(AlterJobV2.JobState.FINISHED_REWRITING, schemaChangeJob.getJobState());

        // Add table back to database
        db.registerTableUnlocked(table);

        // We've mocked ColumnTypeConverter.publishVersion to throw RpcException, should this runFinishedRewritingJob will fail but
        // should not throw any exception.
        schemaChangeJob.runFinishedRewritingJob();
        Assert.assertEquals(AlterJobV2.JobState.FINISHED_REWRITING, schemaChangeJob.getJobState());

        // Make publish version success
        new MockUp<Utils>() {
            @Mock
            public void publishVersion(@NotNull List<Tablet> tablets, TxnInfoPB txnInfo, long baseVersion,
                                       long newVersion, long warehouseId) {
                // nothing to do
            }
        };

        schemaChangeJob.runFinishedRewritingJob();
        Assert.assertEquals(AlterJobV2.JobState.FINISHED, schemaChangeJob.getJobState());
        Assert.assertTrue(schemaChangeJob.getFinishedTimeMs() > System.currentTimeMillis() - 10_000L);

        Assert.assertEquals(2, table.getBaseSchema().size());
        Assert.assertEquals("c0", table.getBaseSchema().get(0).getName());
        Assert.assertEquals("c1", table.getBaseSchema().get(1).getName());

        Assert.assertSame(partition, table.getPartitions().stream().findFirst().get());
        Assert.assertEquals(3, partition.getVisibleVersion());
        Assert.assertEquals(4, partition.getNextVersion());

        shadowIndexes = partition.getMaterializedIndices(MaterializedIndex.IndexExtState.SHADOW);
        Assert.assertEquals(0, shadowIndexes.size());

        List<MaterializedIndex> normalIndexes =
                partition.getMaterializedIndices(MaterializedIndex.IndexExtState.VISIBLE);
        Assert.assertEquals(1, normalIndexes.size());
        MaterializedIndex normalIndex = normalIndexes.get(0);

        // Does not support cancel job in FINISHED state.
        schemaChangeJob.cancel("test");
        Assert.assertEquals(AlterJobV2.JobState.FINISHED, schemaChangeJob.getJobState());
    }