import org.apache.kafka.connect.errors.IllegalWorkerStateException; //导入依赖的package包/类
@Override
public final Iterator<FileMetadata> execute() throws IOException {
    if (hasEnded()) {
        throw new IllegalWorkerStateException("Policy has ended. Cannot be retried");
    }
    preCheck();

    Iterator<FileMetadata> files = Collections.emptyIterator();
    for (FileSystem fs : fileSystems) {
        files = concat(files, listFiles(fs));
    }
    executions.incrementAndGet();

    postCheck();

    return files;
}
 

import org.apache.kafka.connect.errors.IllegalWorkerStateException; //导入依赖的package包/类
@Test(expected = IllegalWorkerStateException.class)
@Override
public void execPolicyAlreadyEnded() throws IOException {
    policy.execute();
    assertFalse(policy.hasEnded());
    policy.interrupt();
    assertTrue(policy.hasEnded());
    policy.execute();
}
 

import org.apache.kafka.connect.errors.IllegalWorkerStateException; //导入依赖的package包/类
@Override
public Set<TopicPartition> assignment() {
    if (consumer == null) {
        throw new IllegalWorkerStateException("SinkTaskContext may not be used to look up partition assignment until the task is initialized");
    }
    return consumer.assignment();
}
 

import org.apache.kafka.connect.errors.IllegalWorkerStateException; //导入依赖的package包/类
@Override
public void pause(TopicPartition... partitions) {
    if (consumer == null) {
        throw new IllegalWorkerStateException("SinkTaskContext may not be used to pause consumption until the task is initialized");
    }
    try {
        for (TopicPartition partition : partitions)
            pausedPartitions.add(partition);
        consumer.pause(Arrays.asList(partitions));
    } catch (IllegalStateException e) {
        throw new IllegalWorkerStateException("SinkTasks may not pause partitions that are not currently assigned to them.", e);
    }
}
 

import org.apache.kafka.connect.errors.IllegalWorkerStateException; //导入依赖的package包/类
@Override
public void resume(TopicPartition... partitions) {
    if (consumer == null) {
        throw new IllegalWorkerStateException("SinkTaskContext may not be used to resume consumption until the task is initialized");
    }
    try {
        for (TopicPartition partition : partitions)
            pausedPartitions.remove(partition);
        consumer.resume(Arrays.asList(partitions));
    } catch (IllegalStateException e) {
        throw new IllegalWorkerStateException("SinkTasks may not resume partitions that are not currently assigned to them.", e);
    }
}
 

import org.apache.kafka.connect.errors.IllegalWorkerStateException; //导入依赖的package包/类
@Test(expected = IllegalWorkerStateException.class)
public void execPolicyAlreadyEnded() throws IOException {
    policy.execute();
    assertTrue(policy.hasEnded());
    policy.execute();
}