flink-user mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Sofer, Tovi " <tovi.so...@citi.com>
Subject RE: Flink kafka consumer that read from two partitions in local mode
Date Mon, 25 Sep 2017 14:17:31 GMT
Hi Gordon,

Thanks for your assistance.


·         We are running flink currently  in local mode(MiniCluster), using flink 1.3.2 and
flink-connector-kafka-0.10_2.10.


·         In Consumer log I see 1 partition only (when parallelism=1), so the problem indeed
seems to be in producer.
2017-09-25 17:10:58,140 WARN  org.apache.kafka.clients.consumer.ConsumerConfig - [Source:
fix_topic -> FixMapper (1/1)] The configuration 'topic.name' was supplied but isn't a known
config.
2017-09-25 17:10:58,143 INFO  org.apache.kafka.common.utils.AppInfoParser - [Source: fix_topic
-> FixMapper (1/1)] Kafka version : 0.10.2.1
2017-09-25 17:10:58,144 INFO  org.apache.kafka.common.utils.AppInfoParser - [Source: fix_topic
-> FixMapper (1/1)] Kafka commitId : e89bffd6b2eff799
2017-09-25 17:10:58,679 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09
- [Source: fix_topic -> FixMapper (1/1)] Got 1 partitions from these topics: [fix]
2017-09-25 17:10:58,679 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer09
- [Source: fix_topic -> FixMapper (1/1)] Consumer is going to read the following topics
(with number of partitions): fix (1),
2017-09-25 17:10:58,680 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumerBase
- [Source: fix_topic -> FixMapper (1/1)] Consumer subtask 0 will start reading the following
1 partitions from the committed group offsets in Kafka: [KafkaTopicPartition{topic='fix',
partition=0}]


·         The producer seems to write to one partition only.

internalProducer.topicPartitionsMap and Cluster.Partitions seems to have one partition for
FIX topic.



In producer log each producer start with configuration below:

2017-09-25 17:06:49,596 INFO  org.apache.kafka.clients.producer.ProducerConfig- [Source: random
-> Sink: fixTopicSink (2/2)] ProducerConfig values:

                acks = 1

                batch.size = 16384

                block.on.buffer.full = false

                bootstrap.servers = [localhost:9092]

                buffer.memory = 33554432

                client.id =

                compression.type = none

                connections.max.idle.ms = 540000

                interceptor.classes = null

                key.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

                linger.ms = 0

                max.block.ms = 60000

                max.in.flight.requests.per.connection = 5

                max.request.size = 1048576

                metadata.fetch.timeout.ms = 60000

                metadata.max.age.ms = 300000

                metric.reporters = []

                metrics.num.samples = 2

                metrics.sample.window.ms = 30000

                partitioner.class = class org.apache.kafka.clients.producer.internals.DefaultPartitioner

                receive.buffer.bytes = 32768

                reconnect.backoff.ms = 50

                request.timeout.ms = 30000

                retries = 0

                retry.backoff.ms = 100

                sasl.jaas.config = null

                sasl.kerberos.kinit.cmd = /usr/bin/kinit

                sasl.kerberos.min.time.before.relogin = 60000

                sasl.kerberos.service.name = null

                sasl.kerberos.ticket.renew.jitter = 0.05

                sasl.kerberos.ticket.renew.window.factor = 0.8

                sasl.mechanism = GSSAPI

                security.protocol = PLAINTEXT

                send.buffer.bytes = 131072

                ssl….

                timeout.ms = 30000

                value.serializer = class org.apache.kafka.common.serialization.ByteArraySerializer

And print when starting:

2017-09-25 17:07:46,907 INFO  org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducerBase-
[Source: random -> Sink: fixTopicSink (2/2)] Starting FlinkKafkaProducer (2/2) to produce
into default topic fix


Thanks,
Tovi

From: Tzu-Li (Gordon) Tai [mailto:tzulitai@apache.org]
Sent: יום ב 25 ספטמבר 2017 15:06
To: Sofer, Tovi [ICG-IT]; Fabian Hueske
Cc: user
Subject: RE: Flink kafka consumer that read from two partitions in local mode

Hi Tovi,

Your code seems to be correct, and as Fabian described, you don’t need parallelism of 2
to read 2 partitions; a single parallel instance of the source can read multiple partitions.

I’m not sure what could have possibly gone wrong at the moment from a first look, so I may
need to randomly ask you some questions:

Could you let me know which Flink version you are on?
Also, could you try searching in the log to see if you find consumer logs such as:
“Consumer subtask ... will start reading the following (numPartitions) partitions from ...:
(partition info) "
You can try setting parallelism of the source to 1, and you should see that the subtask is
reading 2 partitions.

From the metrics log it does seem like the consumer has picked up both partitions 0 and 1,
but no records seem to be coming from partition 0. Have you perhaps tried using a non-Flink
consumer, perhaps the simple console consumer, to read the topic, and see if records from
both partitions are consumed properly?

Let me know, I’m sure we can figure this out somehow.

Cheers,
Gordon

On 24 September 2017 at 9:44:28 AM, Sofer, Tovi (tovi.sofer@citi.com<mailto:tovi.sofer@citi.com>)
wrote:
Thank you Fabian.

Fabian, Gordon, am I missing something in consumer setup?
Should I configure consumer in some way to subscribe to two partitions?

Thanks and regards,
Tovi

From: Fabian Hueske [mailto:fhueske@gmail.com]
Sent: יום ג 19 ספטמבר 2017 22:58
To: Sofer, Tovi [ICG-IT]
Cc: user; Tzu-Li (Gordon) Tai
Subject: Re: Flink kafka consumer that read from two partitions in local mode

Hi Tovi,
your code looks OK to me. Maybe Gordon (in CC) has an idea what is going wrong.
Just a side note: you don't need to set the parallelism to 2 to read from two partitions.
A single consumer instance reads can read from multiple partitions.
Best,
Fabian

2017-09-19 17:02 GMT+02:00 Sofer, Tovi <tovi.sofer@citi.com<mailto:tovi.sofer@citi.com>>:
Hi,

I am trying to setup FlinkKafkaConsumer which reads from two partitions in local mode, using
 setParallelism=2.
The producer writes to two partition (as it is shown in metrics report).
But the consumer seems to read always from one partition only.
Am I missing something in partition configuration?

Code:


Producer setup:
Configuration localConfig = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironment(parallelism,
localConfig);

env.setParallelism(2);

String kafkaPort = parameters.get(SimulatorConfig.ParamsEnum.KAFKA_PORT.fullName());

SingleOutputStreamOperator<String> fixMsgSource = env.addSource(srcMsgProvider.getFixMsgSource(),
TypeInformation.of(String.class)).name(sourceGenerationType.getValue());
fixMsgSource.addSink(new FlinkKafkaProducer010<>("localhost:"  + kafkaPort, TOPIC_NAME,
new SimpleStringSchema()))

.name(“fix_topic”);

env.execute(“MsgSimulatorJob”);


Consumer setup:

String topicName = “fix”;
Configuration conf = new Configuration();
StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);

env.setParallelism(2);
env.getConfig().setGlobalJobParameters(configParams); // make parameters available in the
web interface
DeserializationSchema<Tuple2<Long, String>> deserializationSchema = new SimpleStringAndTimestampDeserializationSchema
();
FlinkKafkaConsumer010<Tuple2<Long, String>> kafkaConsumer = new FlinkKafkaConsumer010<>(topicName,
deserializationSchema, kafkaParams.getProperties());

DataStream<Tuple2<Long, String>> fixMessagesStream = env.addSource(kafkaConsumer).name("fix_topic").setParallelism(2);

As you can see in output, only 1 consumer partition seems to be used:
Producer output:
2017-09-19 14:40:45,818 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.ca94a7e34623274bd560c7115f842acf.MsgSimulatorJob.Source:
random -> Sink: fix_topic.1.numRecordsInPerSecond: 0.0
2017-09-19 14:40:45,818 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.ca94a7e34623274bd560c7115f842acf.MsgSimulatorJob.Sink:
fix_topic.1.numRecordsInPerSecond: 19836.033333333333
2017-09-19 14:40:45,818 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.ca94a7e34623274bd560c7115f842acf.MsgSimulatorJob.Sink:
fix_topic.0.numRecordsInPerSecond: 20337.933333333334
2017-09-19 14:40:45,819 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.ca94a7e34623274bd560c7115f842acf.MsgSimulatorJob.Source:
random -> Sink: fix_topic.0.numRecordsInPerSecond: 0.0
Consumer output:
2017-09-19 14:40:45,116 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.KafkaConsumer.select-rate: 1928.1421153709368
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.KafkaConsumer.commit-rate: 0.21623491761449637
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.KafkaConsumer.outgoing-byte-rate: 982.0051413881748
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.KafkaConsumer.sync-rate: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.KafkaConsumer.io-ratio: 0.01148712465103046
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.numRecordsOutPerSecond: 6625.266666666666
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.numRecordsInPerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.numBytesOutPerSecond: 1.40222884E7
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numRecordsInPerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numBytesInRemotePerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numRecordsInPerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numBytesOutPerSecond: 10.5
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numBytesInLocalPerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numRecordsOutPerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.1.numBytesInRemotePerSecond: 0.0
2017-09-19 14:40:45,117 INFO  - [Flink-MetricRegistry-1] 127.0.0.1.taskmanager.7e87ec82a8abb8ee76e43ad29f682291.Flink
Streaming Job.Source: fix_topic.0.numRecordsOutPerSecond: 0.0


Thanks and regards,
Tovi

Mime
View raw message