flink 写入es失败导致数据丢失

[sunfulin]

hi，
我们现在使用flink消费kafka数据写到es，今天发现在默认设置下，es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。

[Congxian Qiu]

Hi

你 ES Sink 是自己写的，还是用的社区的呢？社区的使用了哪个版本，以及配置是啥样的呢

Best,
Congxian


sunfulin <[hidden email]> 于2020年7月10日周五 上午10:51写道：

> hi，
>
> 我们现在使用flink消费kafka数据写到es，今天发现在默认设置下，es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。

[sunfulin]

hi
我使用社区默认的ES，主要配置如下：我使用flink 1.10.1，blink-planner。使用了ES6的sink。
我看了下文档，默认有个参数是 connector.failure-handler,是fail。我也能在TM日志里看到连接es失败的报错，但是整个任务checkpoint并没有失败。数据丢了。


WITH (
'connector.type' = 'elasticsearch',
'connector.version' = '<ES_YUNTU.VERSION>',
'connector.hosts' = '<ES_YUNTU.SERVERS>',
'connector.index' = 'realtime_fund_product_all_sell',
'connector.document-type' = '_doc',
'update-mode' = 'upsert',
'connector.key-delimiter' = '$',
'connector.key-null-literal' = 'n/a',
'connector.bulk-flush.interval' = '1000',
'format.type' = 'json'
)

















在 2020-07-10 11:00:53，"Congxian Qiu" <[hidden email]> 写道：

>Hi
>
>你 ES Sink 是自己写的，还是用的社区的呢？社区的使用了哪个版本，以及配置是啥样的呢
>
>Best,
>Congxian
>
>
>sunfulin <[hidden email]> 于2020年7月10日周五 上午10:51写道：
>
>> hi，
>>
>> 我们现在使用flink消费kafka数据写到es，今天发现在默认设置下，es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。

[Leonard Xu]

Hello, fulin

> es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。

es 服务挂掉是指es 集群不可用吗？那这时应该是写入es应该失败，作业也会失败，你说的没有积压是指什么呢？

Best
Leonard Xu

[sunfulin]

hi，Leonard
是的。es集群服务不可用。我能观察到写入es失败，但是作业确实没失败。等到es集群服务恢复后，作业也正常了，但是故障期间的数据有丢失。














在 2020-07-10 11:16:17，"Leonard Xu" <[hidden email]> 写道：
>Hello, fulin
>
>> es服务挂掉时，这段时间写入es失败，但是没有积压，而是数据丢失了。这个应该不符合预期。想问下可能是啥原因造成的。
>
>es 服务挂掉是指es 集群不可用吗？那这时应该是写入es应该失败，作业也会失败，你说的没有积压是指什么呢？
>
>Best
>Leonard Xu

[Leonard Xu]

Hi，
我理解作业应该失败才对，你本地可以复现吗？可以复现的话可以在社区建个issue。

Best,
Leonard Xu

> 在 2020年7月10日，11:20，sunfulin <[hidden email]> 写道：
>
> ，但是作业确实没失败。

[Congxian Qiu]

Hi

从官方文档的配置[1] 来看，对于 handle failure 来说，默认是 fail，也就是说 request 失败了会导致作业失败的，可以尝试在
log 中看能否找到这个日志，或者显示的设置成 fail 看看。如果发现 handle failure 是 fail 的情况下不符合预期，可以想
Leonard 说的那样建立一个 issue 来追踪这个问题

[1]
https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/elasticsearch.html#connector-options
Best,
Congxian


Leonard Xu <[hidden email]> 于2020年7月10日周五 上午11:33写道：

> Hi，
> 我理解作业应该失败才对，你本地可以复现吗？可以复现的话可以在社区建个issue。
>
> Best,
> Leonard Xu
>
> > 在 2020年7月10日，11:20，sunfulin <[hidden email]> 写道：
> >
> > ，但是作业确实没失败。
>
>

[sunfulin]

hi,
感谢两位的回复。我先来本地复现下。如果有问题的话，会建个issue。











在 2020-07-10 11:43:33，"Congxian Qiu" <[hidden email]> 写道：

>Hi
>
>从官方文档的配置[1] 来看，对于 handle failure 来说，默认是 fail，也就是说 request 失败了会导致作业失败的，可以尝试在
>log 中看能否找到这个日志，或者显示的设置成 fail 看看。如果发现 handle failure 是 fail 的情况下不符合预期，可以想
>Leonard 说的那样建立一个 issue 来追踪这个问题
>
>[1]
>https://ci.apache.org/projects/flink/flink-docs-release-1.11/dev/table/connectors/elasticsearch.html#connector-options
>Best,
>Congxian
>
>
>Leonard Xu <[hidden email]> 于2020年7月10日周五 上午11:33写道：
>
>> Hi，
>> 我理解作业应该失败才对，你本地可以复现吗？可以复现的话可以在社区建个issue。
>>
>> Best,
>> Leonard Xu
>>
>> > 在 2020年7月10日，11:20，sunfulin <[hidden email]> 写道：
>> >
>> > ，但是作业确实没失败。
>>
>>