https://flink.apache.org/ logo
Join Slack
Powered by
Hi All, We have a data loss issue occurred in prod...
# troubleshooting
s
Hi All, We have a data loss issue occurred in prod. We are using Flink 1.13.2 with Kafka. I am using Flink table APIs to join the streams from Kafka and AWS KDA to run our jobs. I saw the below error in logs. Can someone please help me to understand the below error? I am trying to simulate the error on my local but no luck. Please note while creating the FLink tables in our jobs we are not using Kafka consumer group.id.
Copy code
2023-01-06T21:10:11.885Z {"applicationARN":"arn:aws:kinesisanalytics:us-east-1:542230711021:application/laap-kda-ue1-con-sr-sr-per-dlvry-metrc-imprsn-prod","applicationVersionId":"15","locationInformation":"org.apache.flink.runtime.io.network.netty.PartitionRequestQueue.handleException(PartitionRequestQueue.java:287)","logger":"org.apache.flink.runtime.io.network.netty.PartitionRequestQueue","message":"Encountered error while consuming partitions","messageSchemaVersion":"1","messageType":"ERROR","threadName":"Flink Netty Server (6121) Thread 0","throwableInformation":"org.apache.flink.shaded.netty4.io.netty.channel.unix.Errors$NativeIoException: readAddress(..) failed: Connection reset by peer\n"}
j
This looks like a networking issue, potentially related to timeouts in communication between task managers
Have you ensured your data is well distributed amongst subtasks?
s
Thanks @Jeremy Ber. How I can check whether data is distributed evenly within flink tasks? I am not sure if we can control this. Also do you think in the case of the above communication error Can I expect any data loss or flink will recover by itself?
j
If you check in the flink dashboard and click on one of your tasks you should see a tab called subtasks which will show you the data distribution
It would be dictated either by your join key or how the data comes in via your stream. You can rebalance in order to make it more evenly distributed as well
& hard to say about data loss I'm just going off a hunch of what I see in the error. Can always reprocess messages in flink from upstream to avoid any data loss since last checkpoint
s
yeah, As we are using 
restart-strategy	fixed-delay
, I am also thinking job should recover from the last checkpoint and there should not be any data loss.
✅ 1
I am trying to simulate the same use case but no luck. Please suggest any use case where you think there could be chances of data loss, I will try to simulate and test those on my local.
I have already tried couple of use cases. For example, I am using two Kafka servers one for source and second for sink. I tested my use case by shutdown the sink Kafka server. I have sent data to source Kafka server and after an hour I started the sink Kafka server. I didn’t find any data loss, flink job processed the data and sent to sink Kafka server.
j
Data loss could occur if you update your application without a snapshot and your offsets are lost
Although with kafka it can be configured to rely on topic offset in kafka as well
s
I read somewhere that flink is not using Kafka consumer.group.id to read the offset from Kafka. Do you think we need to set this in our code? As of now we are not using this.
j
No you don't necessarily need to it'll use some default one from flink if you don't set. It's not relying on committed offsets
👍 1
s
With this 
Although with kafka it can be configured to rely on topic offset in kafka as well
- Do you mean to configure group.id? or some other configurations?
Also is there a way to read the offset within flink application? Do we have any metrics related to this?
j
Might be a good place to look
s
Thanks @Jeremy Ber, we are using table APIs 
upsert-kakfa
connector, I will check if we have something similar in table APIs.
a
@Shamit Jain were you ever able to replicate or find the issue? Reading through this morning and found this a little concerning
2 Views
Open in Slack
PreviousNext
Powered by