[FLINK-3061] Kafka Consumer is not failing if broker is not available

• Type: Bug
• Status: Resolved
• Priority: Major
• Resolution: Fixed
• Fix Version/s: 1.0.0
• Component/s: Connectors / Kafka
• Assignee: Robert Metzger
• Reporter: Robert Metzger
• Created: 23/Nov/15 16:46
• Updated: 30/Nov/15 15:16
• Resolved: 30/Nov/15 15:16

Description

It seems that the FlinkKafkaConsumer is just logging the errors when trying to get the initial list of partitions for the topic, but its not failing.

The following code ALWAYS runs, even if there is no broker or zookeeper running.

 def main(args: Array[String]) {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val properties = new Properties()
    properties.setProperty("bootstrap.servers", "localhost:9092")
    properties.setProperty("zookeeper.connect", "localhost:2181")
    properties.setProperty("group.id", "test")
    val stream = env
      .addSource(new FlinkKafkaConsumer082[String]("topic", new SimpleStringSchema(), properties))
      .print

    env.execute("Flink Kafka Example")
  }

The runtime consumers are designed to idle when they have no partitions assigned, but there is no check that there are no partitions at all.