Re: Producer Configuration

Posted by Alexey Trenikhun on
URL: http://deprecated-apache-flink-user-mailing-list-archive.369.s1.nabble.com/Producer-Configuration-tp41757p41779.html

They are picked up, otherwise you would not able to write any messages at all. I believe the page you referring is not for displaying Kafka properties (in my case it is empty as well, but Kafka works). Check logs.

---

From: Claude M <[hidden email]>
Sent: Saturday, February 27, 2021 4:00:23 PM
To: Alexey Trenikhun <[hidden email]>
Cc: user <[hidden email]>
Subject: Re: Producer Configuration

 

Yes, the flink job also works in producing messages.  It's just that after a short period of time, it fails w/ a timeout.  That is why I'm trying to set a longer timeout period but it doesn't seem like the properties are being picked up. 

On Sat, Feb 27, 2021 at 1:17 PM Alexey Trenikhun <[hidden email]> wrote:

Can you produce messages using Kafka console producer connect using same properties ?

---

From: Claude M <[hidden email]>
Sent: Saturday, February 27, 2021 8:05 AM
To: Alexey Trenikhun <[hidden email]>
Cc: user <[hidden email]>
Subject: Re: Producer Configuration

 

Thanks for your reply, yes it was specified.  Sorry I forgot to include it:

 properties.setProperty("bootstrap.servers", "localhost:9092");

On Fri, Feb 26, 2021 at 7:56 PM Alexey Trenikhun <[hidden email]> wrote:

I believe bootstrap.servers is mandatory Kafka property, but it looks like you didn’t set it

---

From: Claude M <[hidden email]>
Sent: Friday, February 26, 2021 12:02:10 PM
To: user <[hidden email]>
Subject: Producer Configuration

 

Hello,

 

I created a simple Producer and when the job ran, it was getting the following error: 

Caused by: org.apache.kafka.common.errors.TimeoutException

I read about increasing the request.timeout.ms.   Thus, I added the following properties.  

Properties properties = new Properties();
properties.setProperty("request.timeout.ms", "30000");
properties.setProperty("retries", "20");
DataStream<String> stream = env.addSource(new SimpleStringGenerator());
stream.addSink(new FlinkKafkaProducer<>("flink-test", new SimpleStringSchema(), properties));

However, after the job is submitted, the User Configuration is empty, please see attached.  

Therefore, it seems these properties are taking into effect since I still have the same problem.  

Any help on these issues are appreciated, thanks.