@Override
    public boolean readyToExecute() throws UserException {
        if (checkReadyToExecuteFast()) {
            return true;
        }

        KafkaRoutineLoadJob kafkaRoutineLoadJob = (KafkaRoutineLoadJob) job;
        Map<Integer, Long> latestOffsets = KafkaUtil.getLatestOffsets(kafkaRoutineLoadJob.getBrokerList(),
                kafkaRoutineLoadJob.getTopic(),
                ImmutableMap.copyOf(kafkaRoutineLoadJob.getConvertedCustomProperties()),
                new ArrayList<>(partitionIdToOffset.keySet()), warehouseId);
        for (Map.Entry<Integer, Long> entry : latestOffsets.entrySet()) {
            kafkaRoutineLoadJob.setPartitionOffset(entry.getKey(), entry.getValue());
        }

        for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
            int partitionId = entry.getKey();
            Long latestOffset = latestOffsets.get(partitionId);
            Long consumeOffset = entry.getValue();
            if (latestOffset != null) {
                if (latestOffset > consumeOffset) {
                    this.latestPartOffset = latestOffsets;
                    return true;
                } else if (latestOffset < consumeOffset) {
                    throw new RoutineLoadPauseException(
                            ERR_ROUTINE_LOAD_OFFSET_INVALID.formatErrorMsg(consumeOffset, latestOffset, partitionId));
                }
            }
        }

        return false;
    }
@Test
    public void testReadyToExecute(@Injectable KafkaRoutineLoadJob kafkaRoutineLoadJob) throws Exception {
        new MockUp<RoutineLoadMgr>() {
            @Mock
            public RoutineLoadJob getJob(long jobId) {
                return kafkaRoutineLoadJob;
            }
        };

        new MockUp<KafkaUtil>() {
            @Mock
            public Map<Integer, Long> getLatestOffsets(String brokerList, String topic,
                                                       ImmutableMap<String, String> properties,
                                                       List<Integer> partitions,
                                                       long warehouseId) throws UserException {
                Map<Integer, Long> offsets = Maps.newHashMap();
                offsets.put(0, 100L);
                offsets.put(1, 100L);
                return offsets;
            }
        };

        Map<Integer, Long> offset1 = Maps.newHashMap();
        offset1.put(0, 99L);
        KafkaTaskInfo kafkaTaskInfo1 = new KafkaTaskInfo(UUID.randomUUID(),
                kafkaRoutineLoadJob,
                System.currentTimeMillis(),
                System.currentTimeMillis(),
                offset1,
                Config.routine_load_task_timeout_second * 1000);
        Assert.assertEquals("kafka", kafkaTaskInfo1.dataSourceType());
        Assert.assertTrue(kafkaTaskInfo1.readyToExecute());

        Map<Integer, Long> offset2 = Maps.newHashMap();
        offset2.put(0, 100L);
        KafkaTaskInfo kafkaTaskInfo2 = new KafkaTaskInfo(UUID.randomUUID(),
                kafkaRoutineLoadJob,
                System.currentTimeMillis(),
                System.currentTimeMillis(),
                offset2,
                Config.routine_load_task_timeout_second * 1000);
        Assert.assertFalse(kafkaTaskInfo2.readyToExecute());

        // consume offset > latest offset
        Map<Integer, Long> offset3 = Maps.newHashMap();
        offset3.put(0, 101L);
        KafkaTaskInfo kafkaTaskInfo3 = new KafkaTaskInfo(UUID.randomUUID(),
                kafkaRoutineLoadJob,
                System.currentTimeMillis(),
                System.currentTimeMillis(),
                offset3,
                Config.routine_load_task_timeout_second * 1000);
        ExceptionChecker.expectThrowsWithMsg(RoutineLoadPauseException.class,
                "Consume offset: 101 is greater than the latest offset: 100 in kafka partition: 0. " +
                        "You can modify 'kafka_offsets' property through ALTER ROUTINE LOAD and RESUME the job",
                () -> kafkaTaskInfo3.readyToExecute());
    }