虽然问题是完全相同的-- Flink(kafka连接器)重新运行它关闭之前看到的最后3-9条消息。
我的版本
Flink 1.1.2
Kafka 0.9.0.1
Scala 2.11.7
Java 1.8.0_91
我的代码
import java.util.Properties
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.CheckpointingMode
import org.apache.flink.streaming.connectors.kafka._
import org.apache.flink.streaming.util.serialization._
import org.apache.flink.runtime.state.filesystem._
object Runner {
def main(args: Array[String]): Unit = {
val env = StreamExecutionEnvironment.getExecutionEnvironment
env.enableCheckpointing(500)
env.setStateBackend(new FsStateBackend("file:///tmp/checkpoints"))
env.getCheckpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE)
val properties = new Properties()
properties.setProperty("bootstrap.servers", "localhost:9092");
properties.setProperty("group.id", "testing");
val kafkaConsumer = new FlinkKafkaConsumer09[String]("testing-in", new SimpleStringSchema(), properties)
val kafkaProducer = new FlinkKafkaProducer09[String]("localhost:9092", "testing-out", new SimpleStringSchema())
env.addSource(kafkaConsumer)
.addSink(kafkaProducer)
env.execute()
}
}
我的SBT依赖项
libraryDependencies ++= Seq(
"org.apache.flink" %% "flink-scala" % "1.1.2",
"org.apache.flink" %% "flink-streaming-scala" % "1.1.2",
"org.apache.flink" %% "flink-clients" % "1.1.2",
"org.apache.flink" %% "flink-connector-kafka-0.9" % "1.1.2",
"org.apache.flink" %% "flink-connector-filesystem" % "1.1.2"
)
我的流程
(3个终端)
TERM-1 start sbt, run program
TERM-2 create kafka topics testing-in and testing-out
TERM-2 run kafka-console-producer on testing-in topic
TERM-3 run kafka-console-consumer on testing-out topic
TERM-2 send data to kafka producer.
Wait for a couple seconds (buffers need to flush)
TERM-3 watch data appear in testing-out topic
Wait for at least 500 milliseconds for checkpointing to happen
TERM-1 stop sbt
TERM-1 run sbt
TERM-3 watch last few lines of data appear in testing-out topic
我的期望
当系统没有错误时,我希望能够在不重新处理已成功完成流的消息的情况下打开和关闭 flink。
我的尝试修复
我添加了对 setStateBackend
的调用,认为默认的内存后端可能没有正确记住。但这似乎没有帮助。
我删除了对 enableCheckpointing
的调用,希望也许在 Flink 和 Zookeeper 中有一个单独的机制来跟踪状态。但这似乎没有帮助。
我使用了不同的 sinks,如 RollingFileSink、print(),希望可能是 kafka 中的 bug。但这似乎没有帮助。
我回滚到 flink(和所有连接器)v1.1.0 和 v1.1.1,希望最新版本中存在 bug。但这似乎没有帮助。
我添加了 zookeeper.connect
配置到属性对象中,希望关于它仅在 0.8 版本中有用的注释是错误的。但这似乎没有帮助。
EXACTLY_ONCE (好主意drfloob),但似乎没有帮助。
我的请求
帮帮我!