首先，这与当我重新运行Flink消费者时，Kafka再次消耗最新消息非常相似，但并不完全相同。那个问题的答案似乎不能解决我的问题。如果我在那个答案中错过了什么，请重新表述答案，因为我显然错过了某些内容。
虽然问题是完全相同的-- Flink（kafka连接器）重新运行它关闭之前看到的最后3-9条消息。
虽然问题是完全相同的-- 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），但似乎没有帮助。
我的请求
帮帮我！