public boolean checkReadyToExecuteFast() {
        KafkaRoutineLoadJob kafkaRoutineLoadJob = (KafkaRoutineLoadJob) job;
        for (Map.Entry<Integer, Long> entry : partitionIdToOffset.entrySet()) {
            int partitionId = entry.getKey();
            Long consumeOffset = entry.getValue();
            Long localLatestOffset = kafkaRoutineLoadJob.getPartitionOffset(partitionId);
            // If any partition has newer data, the task should be scheduled.
            if (localLatestOffset != null && localLatestOffset > consumeOffset) {
                return true;
            }
        }
        return false;
    }
@Test
    public void testCheckReadyToExecuteFast() {
        KafkaRoutineLoadJob kafkaRoutineLoadJob = new KafkaRoutineLoadJob();
        kafkaRoutineLoadJob.setPartitionOffset(0, 101);

        new MockUp<RoutineLoadMgr>() {
            @Mock
            public RoutineLoadJob getJob(long jobId) {
                return kafkaRoutineLoadJob;
            }
        };

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

        Assert.assertTrue(kafkaTaskInfo.checkReadyToExecuteFast());
    }