Flink Streaming example: Kafka010Example.scala doesn't work

[Wojtkowski, Michal]

Hi guys

I'm trying to run official "Kafka010Example.scala", but unortunatelly it doesn't read from input topic and write to output as expected. What am I missing or doing wrong? Any help or hints much appreciated. Here's exactly what I did:

1) Started kafka in docker container (spotify/kafka:latest)

$ docker run -d -p 2181:2181 -p 9092:9092 spotify/kafka:latest

2) Started bash session inside of the container:

$ docker exec -it 26d1cfced4cb /bin/bash

3) Created input and output topics:

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-input

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-output

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --zookeeper localhost:2181 --list

test-input

test-output

4) Launched "Kafka010Example.scala" running flink 1.3.2 in local mode (MiniCluster), with flink-connector-kafka-0.10_2.11

with the following args:

--input-topic test-input --output-topic test-output --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer

5) Logs:

https://file.io/jWsqI8

6) Sent some messages to the topic:

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-input

blah

blahh

blahhh

6) Checked offset on the output topic - NOTHING

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --group myconsumer --topic test-output

Group           Topic                          Pid Offset          logSize         Lag             Owner

myconsumer      test-output                    0   0               0               0               none

7) Check consumer group offset  - NOTHING

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --describe --group myconsumer

No topic available for consumer group provided

GROUP                          TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER

Any clue?

Michal

[Tzu-Li (Gordon) Tai]

Hi Michal,

I can’t seem to access the link you provided for the logs.

As for confirming whether or not some data was read / written, how exactly did you test that?

In the procedure you laid out, it seems like you only performed some consumer group offset checks using the Kafka CLI.

AFAIK, since internally the Kafka consumer does not use the consumer group functionality for partition assignment (instead, static assignment is used), I think the Kafka CLI offset tools will not work with the offsets committed back to Kafka.

On the other hand, the committed offsets are all exposed as the connector’s first-class metrics, which can be queried via the web UI / REST API.

Have you tried to see if the output topic has been written data by simply reading it (e.g. using the Kafka CLI console consumer)?

Cheers,

Gordon

On 15 October 2017 at 8:56:19 PM, Wojtkowski, Michal ([hidden email]) wrote:

Hi guys

I'm trying to run official "Kafka010Example.scala", but unortunatelly it doesn't read from input topic and write to output as expected. What am I missing or doing wrong? Any help or hints much appreciated. Here's exactly what I did:

1) Started kafka in docker container (spotify/kafka:latest)

$ docker run -d -p 2181:2181 -p 9092:9092 spotify/kafka:latest

2) Started bash session inside of the container:

$ docker exec -it 26d1cfced4cb /bin/bash

3) Created input and output topics:

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-input

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-output

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --zookeeper localhost:2181 --list

test-input

test-output

4) Launched "Kafka010Example.scala" running flink 1.3.2 in local mode (MiniCluster), with flink-connector-kafka-0.10_2.11

with the following args:

--input-topic test-input --output-topic test-output --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer

5) Logs:

https://file.io/jWsqI8

6) Sent some messages to the topic:

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-input

blah

blahh

blahhh

6) Checked offset on the output topic - NOTHING

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --group myconsumer --topic test-output

Group           Topic                          Pid Offset          logSize         Lag             Owner

myconsumer      test-output                    0   0               0               0               none

7) Check consumer group offset  - NOTHING

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --describe --group myconsumer

No topic available for consumer group provided

GROUP                          TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER

Any clue?

Michal

[Tzu-Li (Gordon) Tai]

Hi,

Thanks for reporting back! That’s good to know.

Gordon

On 20 October 2017 at 3:51:21 PM, Wojtkowski, Michal ([hidden email]) wrote:

Hi Gordon, 

Thanks for finding time to write back! I managed to solve the issue and it turned out to be entirely related to kafka instance.

Have a good day, cheers!

Michał

On Thu, Oct 19, 2017 at 5:22 PM, Tzu-Li (Gordon) Tai <[hidden email]> wrote:

Hi Michal,

I can’t seem to access the link you provided for the logs.

As for confirming whether or not some data was read / written, how exactly did you test that?

In the procedure you laid out, it seems like you only performed some consumer group offset checks using the Kafka CLI.

AFAIK, since internally the Kafka consumer does not use the consumer group functionality for partition assignment (instead, static assignment is used), I think the Kafka CLI offset tools will not work with the offsets committed back to Kafka.

On the other hand, the committed offsets are all exposed as the connector’s first-class metrics, which can be queried via the web UI / REST API.

Have you tried to see if the output topic has been written data by simply reading it (e.g. using the Kafka CLI console consumer)?

Cheers,

Gordon

On 15 October 2017 at 8:56:19 PM, Wojtkowski, Michal ([hidden email]) wrote:

Hi guys

I'm trying to run official "Kafka010Example.scala", but unortunatelly it doesn't read from input topic and write to output as expected. What am I missing or doing wrong? Any help or hints much appreciated. Here's exactly what I did:

1) Started kafka in docker container (spotify/kafka:latest)

$ docker run -d -p 2181:2181 -p 9092:9092 spotify/kafka:latest

2) Started bash session inside of the container:

$ docker exec -it 26d1cfced4cb /bin/bash

3) Created input and output topics:

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-input

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partitions 1 --topic test-output

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-topics.sh --zookeeper localhost:2181 --list

test-input

test-output

4) Launched "Kafka010Example.scala" running flink 1.3.2 in local mode (MiniCluster), with flink-connector-kafka-0.10_2.11

with the following args:

--input-topic test-input --output-topic test-output --bootstrap.servers localhost:9092 --zookeeper.connect localhost:2181 --group.id myconsumer

5) Logs:

https://file.io/jWsqI8

6) Sent some messages to the topic:

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-console-producer.sh --broker-list localhost:9092 --topic test-input

blah

blahh

blahhh

6) Checked offset on the output topic - NOTHING

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-consumer-offset-checker.sh --zookeeper localhost:2181 --group myconsumer --topic test-output

Group           Topic                          Pid Offset          logSize         Lag             Owner

myconsumer      test-output                    0   0               0               0               none

7) Check consumer group offset  - NOTHING

$ /opt/kafka_2.11-0.10.1.0/bin/kafka-consumer-groups.sh --zookeeper localhost:2181 --describe --group myconsumer

No topic available for consumer group provided

GROUP                          TOPIC                          PARTITION  CURRENT-OFFSET  LOG-END-OFFSET  LAG             OWNER

Any clue?

Michal

--