0

我有两个来源，一个是 Kafka 来源，一个是自定义来源，我需要制作一个睡眠自定义来源一小时，但我遇到了中断。

java.lang.InterruptedException: sleep interrupted
    at java.lang.Thread.sleep(Native Method)
    at com.hulu.hiveIngestion.HiveAddPartitionThread.run(HiveAddPartitionThread.java:48)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:100)
    at org.apache.flink.streaming.api.operators.StreamSource.run(StreamSource.java:63)
    at org.apache.flink.streaming.runtime.tasks.SourceStreamTask$LegacySourceFunctionThread.run(SourceStreamTask.java:201)

代码：

<kafka_Source>.union(<custom_source>)

public class custom_source implements SourceFunction<String> {
public void run(SourceContext<String> ctx)  {
 while(true)
 {
  Thread.sleep(1000);
  ctx.collect("string");
 }
}
}

如何在 Kafka 源将继续其流时制作睡眠自定义源。为什么我得到线程中断异常？

4

1 回答 1

2

这更像是一个 Java 而非 Flink 问题。简而言之，你永远不能依赖 Thread.sleep(x) 来休眠 x 毫秒。正确支持中断也很重要，否则您将无法正常关闭您的工作。

public class custom_source implements SourceFunction<String> {
    private static final Duration SLEEP_DURATION = Duration.ofHours(1);
    private volatile boolean isCanceled = false;

    public void run(SourceContext<String> ctx) {
        while (!isCanceled) {
            // 1 hour wait time
            LocalTime end = LocalTime.now().plusHours(1);
            // this loop ensures that random interruption is not prematurely closing the source
            while (LocalTime.now().compareTo(end) < 0) {
                try {
                    Thread.sleep(Duration.between(LocalTime.now(), end).toMillis());
                } catch (InterruptedException e) {
                    // swallow interruption unless source is canceled
                    if (isCanceled) {
                        Thread.interrupted();
                        return;
                    }
                }
            }
            ctx.collect("string");
        }
    }

    @Override
    public void cancel() {
        isCanceled = true;
    }
}
于 2020-07-30T11:22:18.880 回答