public List<String> getShowInfo() {
        Database db = GlobalStateMgr.getCurrentState().getDb(dbId);
        Table tbl = null;
        if (db != null) {
            Locker locker = new Locker();
            locker.lockDatabase(db, LockType.READ);
            try {
                tbl = db.getTable(tableId);
            } finally {
                locker.unLockDatabase(db, LockType.READ);
            }
        }

        readLock();
        try {
            List<String> row = Lists.newArrayList();
            row.add(String.valueOf(id));
            row.add(name);
            row.add(TimeUtils.longToTimeString(createTimestamp));
            row.add(TimeUtils.longToTimeString(pauseTimestamp));
            row.add(TimeUtils.longToTimeString(endTimestamp));
            row.add(db == null ? String.valueOf(dbId) : db.getFullName());
            row.add(tbl == null ? String.valueOf(tableId) : tbl.getName());
            if (state == JobState.RUNNING) {
                row.add(substate == JobSubstate.STABLE ? state.name() : substate.name());
            } else {
                row.add(state.name());
            }
            row.add(dataSourceType.name());
            row.add(String.valueOf(getSizeOfRoutineLoadTaskInfoList()));
            row.add(jobPropertiesToJsonString());
            row.add(dataSourcePropertiesJsonToString());
            row.add(customPropertiesJsonToString());
            row.add(getStatistic());
            row.add(getProgress().toJsonString());
            row.add(getTimestampProgress().toJsonString());
            switch (state) {
                case PAUSED:
                    row.add(pauseReason == null ? "" : pauseReason.toString());
                    break;
                case CANCELLED:
                    row.add(cancelReason == null ? "" : cancelReason.toString());
                    break;
                case RUNNING:
                    if (substate == JobSubstate.UNSTABLE) {
                        row.add(stateChangedReason == null ? "" : stateChangedReason.toString());
                    } else {
                        row.add("");
                    }
                    break;
                default:
                    row.add("");
            }
            // tracking url
            if (!errorLogUrls.isEmpty()) {
                row.add(Joiner.on(", ").join(errorLogUrls));
                row.add("select tracking_log from information_schema.load_tracking_logs where job_id=" + id);
            } else {
                row.add("");
                row.add("");
            }
            row.add(otherMsg);

            if (RunMode.getCurrentRunMode() == RunMode.SHARED_DATA) {
                try {
                    Warehouse warehouse = GlobalStateMgr.getCurrentState().getWarehouseMgr().getWarehouse(warehouseId);
                    row.add(warehouse.getName());
                } catch (Exception e) {
                    row.add(e.getMessage());
                }
            }
            row.add(getSourceProgressString());

            return row;
        } finally {
            readUnlock();
        }
    }
@Test
    public void testGetShowInfo() throws UserException {
        {
            // PAUSE state
            KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();
            Deencapsulation.setField(routineLoadJob, "state", RoutineLoadJob.JobState.PAUSED);
            ErrorReason errorReason = new ErrorReason(InternalErrorCode.INTERNAL_ERR,
                    TransactionState.TxnStatusChangeReason.OFFSET_OUT_OF_RANGE.toString());
            Deencapsulation.setField(routineLoadJob, "pauseReason", errorReason);

            List<String> showInfo = routineLoadJob.getShowInfo();
            Assert.assertEquals(true, showInfo.stream().filter(entity -> !Strings.isNullOrEmpty(entity))
                    .anyMatch(entity -> entity.equals(errorReason.toString())));
        }

        {
            // Progress
            KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();

            Map<Integer, Long> partitionOffsets = Maps.newHashMap();
            partitionOffsets.put(Integer.valueOf(0), Long.valueOf(1234));
            KafkaProgress kafkaProgress = new KafkaProgress(partitionOffsets);
            Deencapsulation.setField(routineLoadJob, "progress", kafkaProgress);

            Map<Integer, Long> partitionOffsetTimestamps = Maps.newHashMap();
            partitionOffsetTimestamps.put(Integer.valueOf(0), Long.valueOf(1701411708410L));
            KafkaProgress kafkaTimestampProgress = new KafkaProgress(partitionOffsetTimestamps);
            Deencapsulation.setField(routineLoadJob, "timestampProgress", kafkaTimestampProgress);

            routineLoadJob.setPartitionOffset(0, 12345);

            List<String> showInfo = routineLoadJob.getShowInfo();
            Assert.assertEquals("{\"0\":\"12345\"}", showInfo.get(20));
            //The displayed value is the actual value - 1
            Assert.assertEquals("{\"0\":\"1233\"}", showInfo.get(14));
            Assert.assertEquals("{\"0\":\"1701411708409\"}", showInfo.get(15));
        }

        {
            // UNSTABLE substate
            KafkaRoutineLoadJob routineLoadJob = new KafkaRoutineLoadJob();

            Map<Integer, Long> partitionOffsetTimestamps = Maps.newHashMap();
            partitionOffsetTimestamps.put(Integer.valueOf(0), Long.valueOf(1701411708410L));
            KafkaProgress kafkaTimestampProgress = new KafkaProgress(partitionOffsetTimestamps);
            Deencapsulation.setField(routineLoadJob, "timestampProgress", kafkaTimestampProgress);

            routineLoadJob.updateState(RoutineLoadJob.JobState.RUNNING, null, false);
            // The job is set unstable due to the progress is too slow.
            routineLoadJob.updateSubstate();

            List<String> showInfo = routineLoadJob.getShowInfo();
            Assert.assertEquals("UNSTABLE", showInfo.get(7));
            // The lag [xxx] of partition [0] exceeds Config.routine_load_unstable_threshold_second [3600]
            Assert.assertTrue(showInfo.get(16).contains(
                    "partition [0] exceeds Config.routine_load_unstable_threshold_second [3600]"));

            partitionOffsetTimestamps.put(Integer.valueOf(0), Long.valueOf(System.currentTimeMillis()));
            kafkaTimestampProgress = new KafkaProgress(partitionOffsetTimestamps);
            Deencapsulation.setField(routineLoadJob, "timestampProgress", kafkaTimestampProgress);
            // The job is set stable due to the progress is kept up.
            routineLoadJob.updateSubstate();
            showInfo = routineLoadJob.getShowInfo();
            Assert.assertEquals("RUNNING", showInfo.get(7));
            Assert.assertEquals("", showInfo.get(16));

            // The job is set stable.
            routineLoadJob.updateSubstateStable();
            showInfo = routineLoadJob.getShowInfo();
            Assert.assertEquals("RUNNING", showInfo.get(7));
            Assert.assertEquals("", showInfo.get(16));
        }
    }