Mailing-List: contact user-help@flink.apache.org; run by ezmlm
Precedence: bulk
Content-Type: text/plain; charset=utf-8
Mime-Version: 1.0 (Mac OS X Mail 10.3 \(3273\))
Subject: Re: Beam Application run on cluster setup (Kafka+Flink)
From: Piotr Nowojski <piotr@data-artisans.com>
In-Reply-To: <1507816161074-0.post@n4.nabble.com>
Date: Thu, 12 Oct 2017 16:43:44 +0200
Cc: user@flink.apache.org
Content-Transfer-Encoding: quoted-printable
Message-Id: <221CCD52-8850-41EA-B772-98ADFE616342@data-artisans.com>
References: <1507816161074-0.post@n4.nabble.com>
To: Shankara <shankara.odl@gmail.com>
archived-at: Thu, 12 Oct 2017 14:43:56 -0000

Hi,

What do you mean by:

> With standalone beam application kafka can receive the message, But in
cluster setup it is not working.

In your example you are reading the data from Kafka and printing them to =
console. There doesn=E2=80=99t seems to be anything that writes back to =
Kafka, so what do you mean by =E2=80=9CKafka can not receive the =
message=E2=80=9D?

Did you check the output file of your application in the log directory? =
Did you check Flink logs if there are any errors?

Piotrek

> On 12 Oct 2017, at 15:49, Shankara <shankara.odl@gmail.com> wrote:
>=20
> Below is my setup=20
>        1. Kafka zookeeper and server in one machine (192.168.1.116) =
and
> producer (192.168.1.100) and consumer (192.168.1.117) in another =
machine. =20
> --> This work fine no issue=20
>        2. Running standalone beam application with kafka consumer --> =
This
> work fine
>        3. Running beam application in flink cluster with kafka =
consumer -->
> This is not working
>  Not receiving message from kafka producer.
>=20
> Same program works fine with standalone with flink runner.
> Below is my code snippet.
>=20
> public static void main(String[] args) {
>    Pipeline p =3D initializePipeline(args);
>    Map<String, List&lt;String>> intelliOmIms =3D new TreeMap<>();
>=20
>    PTransform<PBegin, PCollection&lt;KV&lt;Integer, byte[]>>> reader;
>    reader =3D KafkaIO.<Integer, byte[]>read()
>            .withBootstrapServers("192.168.1.116:9092")    --->Kafka
> zookeeper and server running
>            .withTopic("kafkatest")
>            .withKeyDeserializer(IntegerDeserializer.class)
>            .withValueDeserializer(IntelliOmImsKpiDataUtil.class)
>            .withoutMetadata();
>=20
>    PCollection<KV&lt;Integer, byte[]>> output =3D p.apply(reader);
>    output.apply(ParDo.of(new PrintMsg()));
>=20
>    p.run().waitUntilFinish();
> }
>=20
>  In IntelliOmImsKpiDataUtil deserializer I am just printing message =
saying
> that kafka is received the message.
>=20
> public static class PrintMsg extends DoFn<KV&lt;Integer, byte[]>, =
Void> {
>=20
>    @ProcessElement
>    public void processElement(ProcessContext c) {
>        System.out.println("Received Message .... from kafkatest Topic =
");
>    }
> }
>=20
>  Started Zookeeper in 192.168.1.116 like below :
>    bin/zookeeper-server-start.sh config/zookeeper.properties
>=20
>  Started Server in 192.168.1.116 like below :
>    bin/kafka-server-start.sh config/server.properties
>=20
>  Started Producer in 192.168.1.100 like below :
>    bin/kafka-console-producer.sh --broker-list 192.168.1.116:9092 =
--topic
> kafkatest
>=20
>  Started Consumer in 192.168.1.117 like below :
>    bin/kafka-console-consumer.sh --zookeeper 192.168.1.116:2181 =
--topic
> kafkatest --from-beginning
>=20
>   With standalone beam application kafka can receive the message, But =
in
> cluster setup it is not working.
>=20
> Can you please help me to check it.=20
>=20
>=20
>=20
>=20
> --
> Sent from: =
http://apache-flink-user-mailing-list-archive.2336050.n4.nabble.com/