FW: Unable to achieve Flink kafka connector exactly once delivery semantics.

...@flink.apache.org , u...@flink.apache.org Subject: Unable to achieve Flink kafka connector exactly once delivery semantics. Hi Team, I am trying to configure my kafka sink connector with “exactly-once” delivery guarantee, however it’s failing when I run the flink job with this configuration, here is the

flink-kafka-connector 消费时获取不到topic-paitition

kafka connector 开了这个配置, 'properties.allow.auto.create.topics' = 'true' 文档里面也有提到 , 但是开启flinkKafkaComsumer消费一个新的topic时，还是报找不到topic的错误，有大佬帮忙解释一下吗？ 报错如下： 这是我的配置 kafka broker 也开启了自动创建topic的配置

Re:Re: Re: Flink Kafka Sink Error: org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default producerId at version 0

感谢回复，我这边问题已经修复了，修改一下clients的版本到2.4.1就可以了 在 2022-08-26 16:20:27，"Weihua Hu" 写道： >可以尝试升级到 2.5+ > >Best, >Weihua > > >On Thu, Aug 25, 2022 at 6:41 PM gulugulucxg wrote: > >> 您好，集群版本是1.1.1，是挺低的，是这个原因吗，升级到多少合适呢 >> 在 2022-08-25 18:31:06，"Weihua Hu" 写道： >> >kafka 集群的版本是什么呢？看起来是集群版本有点低了

Re: Re: Flink Kafka Sink Error: org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default producerId at version 0

可以尝试升级到 2.5+ Best, Weihua On Thu, Aug 25, 2022 at 6:41 PM gulugulucxg wrote: > 您好，集群版本是1.1.1，是挺低的，是这个原因吗，升级到多少合适呢 > 在 2022-08-25 18:31:06，"Weihua Hu" 写道： > >kafka 集群的版本是什么呢？看起来是集群版本有点低了 > > > >Best, > >Weihua > > > > > >On Thu, Aug 25, 2022 at 3:41 PM gulugulucxg wrote: > > > >> 大佬们好： > >> >

Re:Re: Flink Kafka Sink Error: org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default producerId at version 0

您好，集群版本是1.1.1，是挺低的，是这个原因吗，升级到多少合适呢 在 2022-08-25 18:31:06，"Weihua Hu" 写道： >kafka 集群的版本是什么呢？看起来是集群版本有点低了 > >Best, >Weihua > > >On Thu, Aug 25, 2022 at 3:41 PM gulugulucxg wrote: > >> 大佬们好： >> >> 我这边指定EXACTLY_ONCE写kafka后，任务直接起能起来，但是从savepoint起任务总是失败，kafka-clients版本2.5.0，flink版本及相关依赖版本均为1.12.4， >> >>

Re: Flink Kafka Sink Error: org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default producerId at version 0

kafka 集群的版本是什么呢？看起来是集群版本有点低了 Best, Weihua On Thu, Aug 25, 2022 at 3:41 PM gulugulucxg wrote: > 大佬们好： > > 我这边指定EXACTLY_ONCE写kafka后，任务直接起能起来，但是从savepoint起任务总是失败，kafka-clients版本2.5.0，flink版本及相关依赖版本均为1.12.4， > > 异常如下： > > 2022-08-25 10:42:44 > > org.apache.kafka.common.errors.UnsupportedVersionExc

Flink Kafka Sink Error: org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default producerId at version 0

大佬们好： 我这边指定EXACTLY_ONCE写kafka后，任务直接起能起来，但是从savepoint起任务总是失败，kafka-clients版本2.5.0，flink版本及相关依赖版本均为1.12.4， 异常如下： 2022-08-25 10:42:44 org.apache.kafka.common.errors.UnsupportedVersionException: Attempted to write a non-default producerId at version 0 相关代码如下： Properties properties

Re: Flink Kafka e2e exactly once问题询问

假设我的Checkpoint超时时间15min，kafka transcation超时时间10min，在Checkpoint长时间卡住，超过了事务超时时间，这种情况是不是就有问题了 ？ 赵珩 于2021年12月31日周五 21:23写道： > 我的理解是超过kafka transaction > timeout时间重启flink任务才会发生未提交数据丢失的情况， > kafka不会无限期的保存未提交事务数据。 > 正常情况下的flink重启是不会出现数据丢失的。 > > 在 2021/12/31 11:31, zilong xiao 写道: > > 看官方文档中有介绍说当kafka事务

Re: Flink Kafka e2e exactly once问题询问

kafka是有拒绝超时事务提交机制的 Michael Ran 于2021年12月31日周五 14:40写道： > 没测试过，如果kafka 确定自身会丢掉超时事务消息的前提下，比如10分钟超时丢消息。1.flink > 发送消息A，进入第一阶段。2.flink 等待kafka 消息一阶段 ack信息 3.flink > 收到ack消息，发送二阶段确认消息，并进行chk异常： > 这个时候flink第二阶段消息确认，发送失败（同时flink应用因为各种原因挂了，超过10分钟）3.1 10分钟后，kakfa > 丢弃事务超时的消息3.2 flink 重启，重新提交二阶段的事务id

Re: Flink Kafka e2e exactly once问题询问

我的理解是超过kafka transaction timeout时间重启flink任务才会发生未提交数据丢失的情况， kafka不会无限期的保存未提交事务数据。 正常情况下的flink重启是不会出现数据丢失的。 在 2021/12/31 11:31, zilong xiao 写道: 看官方文档中有介绍说当kafka事务超时时，可能会出现数据丢失的情况，那就是说，Flink没办法完全保证端到端exactly once是么？想请教下社区大佬，我这么理解是正确的吗？一直都听说Flink 写kafka是可以保证端到端exactly once的，看到文档描述有点懵 文档地址： https

Flink Kafka e2e exactly once问题询问

看官方文档中有介绍说当kafka事务超时时，可能会出现数据丢失的情况，那就是说，Flink没办法完全保证端到端exactly once是么？想请教下社区大佬，我这么理解是正确的吗？一直都听说Flink 写kafka是可以保证端到端exactly once的，看到文档描述有点懵 文档地址： https://nightlies.apache.org/flink/flink-docs-release-1.13/docs/connectors/datastream/kafka/#kafka-producers-and-fault-tolerance 关键字："If the time betwee

回复：Flink kafka自定义metrics在influxdb上解析失败

你好，我没有自己开发连接器，我用的是kafka connector，influxdb只是作为一个metrics信息存储端，你是需要一个influxdb的连接器？我只是用到了metrics统计体系而已，只是自定义了Counter对象，和连接器没有什么关系 发自 网易邮箱大师 回复的原邮件 | 发件人 | 信华哺 | | 日期 | 2021年11月26日 17:22 | | 收件人 | user-zh@flink.apache.org | | 抄送至 | | | 主题 | 回复：Flink kafka自定义metrics在influxdb上解析失败 | 你好

回复：Flink kafka自定义metrics在influxdb上解析失败

你好： 我想问一下，你用的flink sql连接器是自己开发的么？ 我在网上只能找到一个datastream的influxdb连接器 在2021年7月23日 10:11，Jimmy Zhang 写道： 大家好，Flink版本1.13.1。 我在FlinkKafkaProducer的invork方法中，创建了一个counter，它采用sinkTableName+_receive_count作为最终的metric 表名，并且采用influxDB作为上报终端，其中，sinkTableName是从catalog信息里拿到的，我从构造中一层层传递进去。 但是发现一个奇怪的问题：当我

回复：Flink kafka自定义metrics在influxdb上解析失败

Hi,caizhi，非常感谢你的回复！ 在KafkaDynamicTableFactory.java的createDynamicTableSink(Context context)方法开始，我通过context.getObjectIdentifier().getObjectName()获取到sinkTableName。因为ObjectIdentifier类就唯一标识了一个表，它包括catalogName、databaseName和objectName,分别代表catalog名、数据库名和表名。之后，我通过构造传入到了FlinkKafkaProducer,然后就可以使用了。 我已经解决了该问

Re: Flink kafka自定义metrics在influxdb上解析失败

Hi！ 是通过什么方式拿到 sink table name 呢？从报错信息来看，拿到的可能并不是“a” 这样的 table name，而是一个 insert 语句的 digest，因此 metric 表名会变得比较复杂，导致解析错误。 当然也可以选择对 metric 表名进行转义。 Jimmy Zhang 于2021年7月23日周五 上午10:11写道： > 大家好，Flink版本1.13.1。 > 我在FlinkKafkaProducer的invork方法中，创建了一个counter，它采用sinkTableName+_receive_count作为最终的metric > 表名，

Flink kafka自定义metrics在influxdb上解析失败

大家好，Flink版本1.13.1。 我在FlinkKafkaProducer的invork方法中，创建了一个counter，它采用sinkTableName+_receive_count作为最终的metric 表名，并且采用influxDB作为上报终端，其中，sinkTableName是从catalog信息里拿到的，我从构造中一层层传递进去。 但是发现一个奇怪的问题：当我采用简单的形如insert into a select from b where b.c='d'的SQL语句时，influxDB中的表可以成功被建出来； 但如果加上UDF，比如 insert into a selec

flink Kafka producer执行出错

我在flink处理消费kafka的一个topic，将迟到数据通过侧流发送到一个新的topic，基本上是一直报这个错，每次提交checkpoint时都会报这个错，然后就会重启 还请指导一下，需要做些其它的设置吗 2021-04-30 17:00:51 org.apache.flink.streaming.connectors.kafka.FlinkKafkaException: Failed to send data to Kafka: Producer attempted an operation with an old epoch. Either there is a newer pr

Re: Flink-kafka-connector Consumer配置警告

flink.partition-discovery.interval-millis这个配置在Flink中是生效的，flink kafka connectors 会根据配置的时间去获取kafka topic的分区信息，代码实现见： FlinkKafkaConsumerBase 中的createAndStartDiscoveryLoop方法。 19:38:37,557 WARN org.apache.kafka.clients.consumer.ConsumerConfig [] - The configuration 'flink.part

回复： Flink-kafka-connector Consumer配置警告

你可以看下源码： 这个props只是作为FlinkKafkaConsumer初始化配置变量，只是这个props 不仅仅是用来初始化kafka的，只不过这个props最后整个扔进kafka消费客户端的初始化里面而已，不会有任何影响。 就想你自己初始化一个kafka 消费端，你往props塞进其他参数，也会警告，但没有任何影响。 -- 原始邮件 -- 发件人:

Re: Flink-kafka-connector Consumer配置警告

这个是 Kafka client 的警告。这个配置项是 Flink 加进去的，Kafka 不认识。 Best, Paul Lam > 2021年4月18日 19:45，lp <973182...@qq.com> 写道： > > flink1.12正常程序中，有如下告警： > > 19:38:37,557 WARN org.apache.kafka.clients.consumer.ConsumerConfig > > [] - The configuration 'flink.partition-discovery.interval-millis' was

Flink-kafka-connector Consumer配置警告

flink1.12正常程序中，有如下告警： 19:38:37,557 WARN org.apache.kafka.clients.consumer.ConsumerConfig [] - The configuration 'flink.partition-discovery.interval-millis' was supplied but isn't a known config. 我有一行如下配置： properties.setProperty(FlinkKafkaConsumerBase.KEY_PARTITION_DISCOVERY_INTERVAL_

flink-kafka-connector Producer.setWriteTimestampToKafka(true) 导致的问题

我使用flink1.12版本，采用flink-kafka-connector从kafka的topicA中读取数据，然后sink会topicB，在sink to topicB的FlinkProducer设置如下时，程序会偶现报错，去掉后异常消失，请问是什么原因呢？ flinkKafkaProducer.setWriteTimestampToKafka(true); -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink kafka connector 偶现报错 Permission denied: connect

我写了一个flink kafka connector的作业，从kafka topicA消费数据，做处理后，又sink回 topicB, 程序正常running中，偶现如下报错： java.net.SocketException: Permission denied: connect at sun.nio.ch.Net.connect0(Native Method) ~[?:1.8.0_231] at sun.nio.ch.Net.connect(Net.java:454) ~[?:1.8.0_231] at

??????????????????flink ????????????????????????kafka,mysql??

??         ??flink ??

flink-Kafka 报错：ByteArraySerializer is not an instance of org.apache.kafka.common.serialization.Serializer

测试代码如下： -- public class Sink_KafkaSink_1{ public static void main(String[] args) throws Exception { final ParameterTool params = ParameterTool.fromPropertiesFile(Sink_KafkaSink_1.class.getResourceAsStream("/pro.properties")); String host = params.get("hos

??????flink??????kafka offset

savepoint??flink-kafka-connector?? Kafka https://ci.apache.org/projects/flink/flink-docs-release-1.12/zh/dev/connectors/kafka.html | | ?? | | 18500348...@163.com | ?? ??2021??1??16?? 19:38

flink??????kafka offset

?? flink ??checkpoint,kafka offsetstate,checkpointingoffset??kafka,??kafka consumer_topic?? groupid??offset,??bakup checkpoint??-s ??chkpoint/save

Re: flink-kafka-sink

hi，没有效果 具体是啥？ cxx <1156531...@qq.com> 于2021年1月7日周四 上午9:53写道： > 我从kafka消费一条数据，然后将消息进行切分，再发送到下游的kafka中，但是这样不能保证在一个事务里面。 > 例如：我将一条数据切分成10条，然后再第五条的时候抛出一个异常，但是前四条已经发送到下游的kafka了。 > 我设置了事务id，隔离级别，client > id，enable.idempotence,max.in.flight.requests.per.connection,retries > 但是没有效果。 > > > > -- > Sen

flink-kafka-sink

我从kafka消费一条数据，然后将消息进行切分，再发送到下游的kafka中，但是这样不能保证在一个事务里面。 例如：我将一条数据切分成10条，然后再第五条的时候抛出一个异常，但是前四条已经发送到下游的kafka了。 我设置了事务id，隔离级别，client id，enable.idempotence,max.in.flight.requests.per.connection,retries 但是没有效果。 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Can flink kafka producer dynamically discover new parititons after expansion??

parititon after I expand the Kafka topic's partition number. but sink kafka don't work like this?? The flink kafka producer get topic's paritition list and cache in topicPartitionsMap as showed In Class FlinkKafkaProducer

Can flink kafka producer dynamically discover new parititons after expansion??

parititon after I expand the Kafka topic's partition number. but sink kafka don't work like this?? The flink kafka producer get topic's paritition list and cache in topicPartitionsMap as showed In Class FlinkKafkaProducer

Re: Flink Kafka作业异常重启，从Checkpoint恢复后重复推送数据部分数据如何处理？

谢谢您的回复，刚也看到官方文档DataStream的Kafka连接器中也有相关说明。 -- Sent from: http://apache-flink.147419.n8.nabble.com/

Re: Flink Kafka作业异常重启，从Checkpoint恢复后重复推送数据部分数据如何处理？

Hi checkpoint 只能保证 state 的 exactly once，但是单条数据可能重复处理多次，如果是 sink 输出多次的话，或许你可以看一下 TwoPhaseCommitSinkFunction 相关的，这篇文章有一个相关的描述[1] [1] https://www.ververica.com/blog/end-to-end-exactly-once-processing-apache-flink-apache-kafka Best, Congxian Kevin Dai <154434...@qq.com> 于2020年8月28日周五 上午9:44写道： >

Flink Kafka作业异常重启，从Checkpoint恢复后重复推送数据部分数据如何处理？

Flink ETL作业生成实时DWD宽表数据，写入Kafka中。 当ETL作业的TM出现异常，自动重启恢复后，作业虽然能从上一次Checkpoint状态恢复，但是会出现重复推送部分数据，导致下游DWS相关作业都要进行去重处理，增加下游作业成本。 想了下解决方案，扩展Kafka Sink，初始化的时候，先读取当前State中记录的位置后面的所有数据，然后写入的时候进行去重处理，恢复到正常位置后，清理掉这部分数据。 想问下大佬们，这种处理方式是否合理，或者有没其他更好的解决方案？ -- Sent from: http://apache-flink.147419.n8.nabble.com/

flink kafka 分区问题

大佬们，有个问题一直不理解 1.FlinkKafkaProducer 往 kafka 中写数据时 kafka 有10分区，现在使用 round-robin 的方式进行分区，每个 task 都会轮循的写下游的所有 partition 为什么10分区的产生的偏移量会出现偏差。 2.多流join 数据偏移，如何修复 3.flink sql 能否repartition

Re: flink kafka 数据丢失

问题123等其他人，问题4：我感觉本质上区别不大，如果你资源也完全对等的话，比如最终总体分配内存量相同。而且也没必要启10个taskmanager，唯一好处是更加隔离，比如某个taskmanager失败不会导致全部slot失败。但感觉10个tm的方式从linux角度估计会稍微浪费部分资源。 steven chen 于2020年8月20日周四 上午9:23写道： > hi： >版本：flink 1.10 +kafka +hive catalog > > >现在我是使用flinkSql ddl 在hivecatalog 注册元数据表，现在启动了4个job ，kafka 4个

flink kafka 数据丢失

hi： 版本：flink 1.10 +kafka +hive catalog 现在我是使用flinkSql ddl 在hivecatalog 注册元数据表，现在启动了4个job ，kafka 4个分区，然后4job 分别根据自己需求去根据hivecatalog 元数据表进行统计，但是出现其中2个job 丢了1条数据，查看kafka 分区偏移量丢失数据分区的偏移量少1。相当于所有任务都共用一个元数据，但是各个统计的任务比如天统计和时统计最后的总数发生偏移，比如天统计为50，时统计则51 | Partition | Latest Offset | Leader | Rep

??????????: Re: Flink????Kafka??Mysql?? End-To-End Exactly-Once????????????

MySQL??Connection?? ??Connection A T ??commit()Connection A??A TwoPhaseCommitSinkFunction pendingCommitTransactions??Connection BcommitT

Re: flink kafka SQL Connectors 传递kerberos 参数

不知道你的问题是能否通过这个解决 我看了下目前文档里缺少了传递kafka properties 的部分，我建了个issue[1]把文档补齐 Best Leonard [1] https://issues.apache.org/jira/browse/FLINK-18768 > 在 2020年7月30日，17:52，lydata 写道： > > > > > > > > 谢谢 ，我试试 > > > > > > > > > > > > > >

Re:Re: flink kafka SQL Connectors 传递kerberos 参数

谢谢 ，我试试 在 2020-07-30 17:34:41，"Leonard Xu" 写道： >Hi, >kafka properties 的参数是可以透传的，你试试下面： > >‘properties.security.protocol'='SASL_PLAINTEXT', >‘properties.sasl.mechanism'='GSSAPI’, >‘properties.sasl.kerberos.service.name'='kafka', > >祝好 >Leonard > > >> 在 2020年7月30日，17:00，lydata

Re: flink kafka SQL Connectors 传递kerberos 参数

Hi, kafka properties 的参数是可以透传的，你试试下面： ‘properties.security.protocol'='SASL_PLAINTEXT', ‘properties.sasl.mechanism'='GSSAPI’, ‘properties.sasl.kerberos.service.name'='kafka', 祝好 Leonard > 在 2020年7月30日，17:00，lydata 写道： > > > > 是否需要这3个参数，或者下面参数是否支持？ > > > > > 'security.protocol'='SASL_PLA

flink kafka SQL Connectors 传递kerberos 参数

flink v1.11.1 kafka使用了kerberos 下面DDL 是支持 kerberos 参数 CREATETABLEkafkaTable( ... )WITH('connector'='kafka', 'topic'='user_behavior', 'properties.bootstrap.servers'='localhost:9092', 'properties.group.id'='testGroup', 'security.protocol'='SASL_PLAINTEXT', 'sasl.mechanism'='GSSAPI', 'sasl.kerbero

flink????kafka??????????

flink1.11kafkagroup offset??kafka tooloffset??flink

flink????kafka????????????????????

??flinkflinkkafka

Flink ??kafka??????????????????checkpoint??????????

>>>>?? >>>>??Flink ??kafka??checkpointEXACTLY_ONCE >>?? Producer attempted an operation with an old epoch.Either there is a newer producer with the same transactionalId, or the producer's transaction has been expired by the broker >>?? >>??

Re: flink kafka connector中获取kafka元数据

hi、 其实我这面对key和timestamp需求不大，而对topic、partition、offset需求大一点，主要有两种场景： 1、根据offset筛选出一组数据的最新值，这个之所以不用时间戳来筛选是因为同一个时间戳可能对应多条数据 2、这个场景有点像kafka数据的备份，在存储端要存储topic、partition、offset元数据信息 On Tue, Jul 7, 2020 at 6:14 PM Jark Wu wrote: > Hi, > > 社区可能打算先支持 timestamp 和 key 的读写，其他 meta 信息（topic, partition, etc..）

Re: flink kafka connector中获取kafka元数据

Hi, 社区可能打算先支持 timestamp 和 key 的读写，其他 meta 信息（topic, partition, etc..）可能会以后再支持。 你对其他 meta 信息的读取需求大吗？ 能具体说说你的业务场景不？感谢！ Best, Jark On Tue, 7 Jul 2020 at 17:42, Dream-底限 wrote: > 好的 > > On Tue, Jul 7, 2020 at 5:30 PM Leonard Xu wrote: > > > 嗯，这个在FLIP-107里会支持，目前没法拿到这些meta数据，可以关注下FLIP-107的进展。 > >

Re: flink kafka connector中获取kafka元数据

好的 On Tue, Jul 7, 2020 at 5:30 PM Leonard Xu wrote: > 嗯，这个在FLIP-107里会支持，目前没法拿到这些meta数据，可以关注下FLIP-107的进展。 > > Best, > Leonard Xu > > > 在 2020年7月7日，17:26，Dream-底限 写道： > > > > hi > > 是的，想以下面这种方式获取 > > > > CREATE TABLE MyUserTable (key string,topic string,,以及其他的数据字段) WITH > > ('connector.type'

Re: flink kafka connector中获取kafka元数据

嗯，这个在FLIP-107里会支持，目前没法拿到这些meta数据，可以关注下FLIP-107的进展。 Best, Leonard Xu > 在 2020年7月7日，17:26，Dream-底限 写道： > > hi > 是的，想以下面这种方式获取 > > CREATE TABLE MyUserTable (key string,topic string,,以及其他的数据字段) WITH > ('connector.type' = 'kafka','connector.version' = '0.11' ,...) > > > On Tue, Jul 7, 2020 at

Re: flink kafka connector中获取kafka元数据

hi 是的，想以下面这种方式获取 CREATE TABLE MyUserTable (key string,topic string,,以及其他的数据字段) WITH ('connector.type' = 'kafka','connector.version' = '0.11' ,...) On Tue, Jul 7, 2020 at 5:19 PM Leonard Xu wrote: > Hi, > kafka元数据 是指kafka记录里的 meta数据吗？ 比如kafka自带的timestamp，kafka的key信息。 > 如果是这些信息的话， Table/SQL

Re: flink kafka connector中获取kafka元数据

Hi, kafka元数据 是指kafka记录里的 meta数据吗？ 比如kafka自带的timestamp，kafka的key信息。 如果是这些信息的话， Table/SQL API 目前还没办法拿到， FLIP-107[1] 会支持这个事情。 祝好， Leonard Xu [1] https://cwiki.apache.org/confluence/display/FLINK/FLIP-107%3A+Reading+table+columns+from+different+parts+of+source+records

flink kafka connector中获取kafka元数据

hi、 flink table/sql api中，有办法获取kafka元数据吗？ tableEnvironment.sqlUpdate(CREATE TABLE MyUserTable (...) WITH ('connector.type' = 'kafka','connector.version' = '0.11' ,...))

??????flink????kafka????????

Hi ?? dercd_seeme-3 partition kafkapartition Best, Yichao Yang --  -- ??: "??"

flink ???? Kafka ???? eos ????

Hi All??  flink kafka ?? eos ?? ?? 0-05-21 16:52:15,057 INFO org.apache.flink.runtime.executiongraph.ExecutionGraph- Source: Custom Source (1/1) (f65b2869d898a050238c53f9fbc9573b) switched from DEPLOYING to RUNNING

?????? Re: Re: Re: ??????:flink ????kafka source ????????????

lib --  -- ??: "JingsongLee"https://ci.apache.org/projects/flink/flink-docs-master/ops/cli.html#usage -- From:Others <41486...@qq.com> Send Time:2020??

????????????:flink ????kafka source ????????????

lib --  -- ??: "AS"https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/connect.html#dependencies jarflink??lib??() ??.     ??. ??2020??01??15?? 14:59?

????????????:flink ????kafka source ????????????

Hi: ??, kafka??factory. ?? https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/table/connect.html#dependencies jarflink??lib??() ??. ??. ??2020??01??15?? 14:59??Others<41486...@qq.com> ?

??????:flink ????kafka source ????????????

flink ??1.9.1 ?? 2020-01-15 11:57:44,255 ERROR org.apache.flink.runtime.webmonitor.handlers.JarRunHandler- Unhandled exception. org.apache.flink.client.program.ProgramInvocationException: The main method caused an error: findAndCreateTableSour

?????? Re: flink????kafka????????????kafka??????????????????

??kafka??Offset?? kafka-run-class.sh kafka.tools.ConsumerOffsetChecker --zookeeper zkhost:2181 --group ${group.id} --topic ${topic_name} zkhost ??group.id??topic_name Group           Topic                          Pid Offset          logSize         Lag 

??????flink????Kafka????????????

sourcekafka ------ ??: "sunfulin"https://manual.sensorsdata.cn/sa/latest/page-1573828.html??Flink????Kafka??kafka console consumer?? flinkjob??

??????flink????Kafka????????????

?? ?? --  -- ??: "sunfulin"https://manual.sensorsdata.cn/sa/latest/page-1573828.html??Flink????Kafka??kafka consol

??????flink????Kafka????????????

kafkaadvertised.host.name --  -- ??: "sunfulin"https://manual.sensorsdata.cn/sa/latest/page-1573828.html??Flink????Kafka??kafka console consumer??

?????? Flink????Kafka??Mysql?? End-To-End Exactly-Once????????????

??1??mysql-connector-java??latest ??5.1.6?? 5.1.48 https://github.com/lusecond/flink_help --depth=1 > > > TwoPhaseCommitSinkFunction4??beginTransaction??preCommit??commit??abort > ?

flink????kafka??????

Hi all,??flink??topicgroup idkakfaflink??group id

??????Flink ????Kafka Window??????????

??: https://ci.apache.org/projects/flink/flink-docs-release-1.9/dev/event_time.html idling sources?? --  -- ??: ""

Re: flink kafka consumer部分消息未消费

web界面上可以看到每个节点收到了多少条数据，发送了多少条数据，看看在哪里少了。 > 在 2019年9月27日，下午7:44，zenglong chen 写道： > > 一个test topic，手动往里面添加消息，flink消费后print输出，发现会漏掉部分添加的消息，问题可能出在哪里？前辈有人遇到过类似问题吗？

flink kafka consumer部分消息未消费

一个test topic，手动往里面添加消息，flink消费后print输出，发现会漏掉部分添加的消息，问题可能出在哪里？前辈有人遇到过类似问题吗？

????flink??kafka??????????????????

flink on yarn?? flink??1.7.2??hadoop??2.8.5??kafka??1.0.0 kafkaflinkkafka??offset ?? Properties props = new Properties(); props.put("auto.offset.reset", "latest"); Schema(), props));DataStream data

Re: Flink Kafka Connector相关问题

pache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#kafka-connector-metrics > > > On 2019/8/22, 7:21 PM, "戴鑫铉" wrote: > > 您好，这次发邮件主要想请教一下关于Flink Kafka Connector的一些问题： > > > 1、在官方文档中有提到关闭checkpoint，flink仍会定时提交offset。但经过测试和观察源码，发现只有FlinkKafkaConsumer0

Re: Flink Kafka Connector相关问题

table/dev/connectors/kafka.html#kafka-consumers-offset-committing-behaviour-configuration [2] https://ci.apache.org/projects/flink/flink-docs-stable/dev/connectors/kafka.html#kafka-connector-metrics On 2019/8/22, 7:21 PM, "戴鑫铉" wrote: 您好，这次发邮件主要想请教一下关于Flink Kafka Connector

Flink Kafka Connector相关问题

您好，这次发邮件主要想请教一下关于Flink Kafka Connector的一些问题： 1、在官方文档中有提到关闭checkpoint，flink仍会定时提交offset。但经过测试和观察源码，发现只有FlinkKafkaConsumer08会在createFetcher时就创建一个定期提交offset的committer，而FlinkKafkaConsumer09和FlinkKafkaConsumer010似乎只能依赖commitInternalOffsetsToKafka()方法提交Offset，该方法只有在checkpoint打开的情况才会被调用。这是否意味着使用

?????? Re: flink kafka???????????????? ???? taskmanager ????

jarflinkflink -- -- ??: "rockey...@163.com"; : 2019??7??26??(??) 10:22 ??: "user-zh"; : Re: Re: flink kafka?

Re: Re: flink kafka相关任务执行出错 导致 taskmanager 退出

你好，zhisheng， 我们并没有用到 flink-metrics-prometheus，只是任务执行的时候就会报出 找不到 org/apache/kafka/common/metrics/stats/Rate$1 这个类，不知道是否是flink自己调用到的 rockey...@163.com 发件人： zhisheng 发送时间： 2019-07-25 17:27 收件人： user-zh 主题： Re: flink kafka相关任务执行出错 导致 taskmanager 退出 hi，rockeycui 你是使用了 flink-metrics

回复: 回复: flink kafka相关任务执行出错 导致 taskmanager 退出

谢谢，我们这边排查一下 rockey...@163.com 发件人： athlon...@gmail.com 发送时间： 2019-07-25 17:30 收件人： user-zh 主题： 回复: flink kafka相关任务执行出错 导致 taskmanager 退出 看看classpath里有没有kafka-clients jar包,这个类是在这个里面的Rate类没有找到如果有那么看看你使用的kafka版本和kafka-clients是否匹配,我遇到类似的问题由于版本不同导致有的类找不到

回复: flink kafka相关任务执行出错 导致 taskmanager 退出

看看classpath里有没有kafka-clients jar包,这个类是在这个里面的Rate类没有找到如果有那么看看你使用的kafka版本和kafka-clients是否匹配,我遇到类似的问题由于版本不同导致有的类找不到 athlon...@gmail.com 发件人： rockey...@163.com 发送时间： 2019-07-25 17:26 收件人： user-zh 主题： flink kafka相关任务执行出错 导致 taskmanager 退出 你好，flink 运行过程中出现以下异常

Re: flink kafka相关任务执行出错 导致 taskmanager 退出

hi，rockeycui 你是使用了 flink-metrics-prometheus 去 report metrics 数据吗？看到不少和 metrics 相关的字眼 rockey...@163.com 于2019年7月25日周四 下午5:12写道： > 你好，flink 运行过程中出现以下异常，有何建议指导排除异常， > > 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser > - Kafka version : 0.9.0.1 > 2019-

flink kafka相关任务执行出错 导致 taskmanager 退出

你好，flink 运行过程中出现以下异常，有何建议指导排除异常， 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 0.9.0.1 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 23c69d62a0cabf06 2019-07-23 10:52

flink kafka相关任务执行出错 导致 taskmanager 退出

你好，flink 运行过程中出现以下异常，有何建议指导排除异常， 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka version : 0.9.0.1 2019-07-23 10:52:01,420 INFO org.apache.kafka.common.utils.AppInfoParser - Kafka commitId : 23c69d62a0cabf06 2019-07-23 10:52:

flink kafka source在并行分布式下是怎么确定一个subtask消费哪个kafka partition的?

Hi All, 我们写kafka comsumer通常都不需要去管消费哪个partition，comsumer会根据partition.assignment.strategy设置的分配策略自动协商分配每个线程消费哪个或者哪些分区 但在FlinkKafkaConsumer中调用的KafkaConsumerThread这个消费线程代码中有如下代码 try { if (hasAssignedPartitions) { newPartitions = unassignedPartitionsQueue.pollBatch(); } else { /

Re:Re: flink-kafka Trigger 无法触发问题

非常感谢您的答复。接下来按照您提供的思路进行排查 在 2019-05-15 16:23:04，"Terry Wang" 写道： >有可能是并行度设置大时，source的部分并发没有数据，导致eventTime未更新。可以排查下是否是这个问题 > >> 在 2019年5月15日，下午2:18，13341000780 <13341000...@163.com> 写道： >> >> hi, 各位大牛好！ >> 自定义了窗口触发器trigger，在onElement函数中注册了EventTimeTimer。出现了很离奇的问题，当并行度Parallelism >> 设置的比s

Re: flink-kafka Trigger 无法触发问题

有可能是并行度设置大时，source的部分并发没有数据，导致eventTime未更新。可以排查下是否是这个问题 > 在 2019年5月15日，下午2:18，13341000780 <13341000...@163.com> 写道： > > hi, 各位大牛好！ > 自定义了窗口触发器trigger，在onElement函数中注册了EventTimeTimer。出现了很离奇的问题，当并行度Parallelism > 设置的比slots数和CPU核数小时，能成功触发onEventTime函数，当大于slots数或者大于CPU核数时，发现无法触发onEventTime，已确定元素能成功进

flink-kafka Trigger 无法触发问题

hi, 各位大牛好！ 自定义了窗口触发器trigger，在onElement函数中注册了EventTimeTimer。出现了很离奇的问题，当并行度Parallelism 设置的比slots数和CPU核数小时，能成功触发onEventTime函数，当大于slots数或者大于CPU核数时，发现无法触发onEventTime，已确定元素能成功进入窗口，即onElement函数能成功触发。有人遇到过类似的问题吗，求解答。 非常感谢.

Re: flink-kafka 消费组问题

Flink 的Kafka source没有使用 Kafka 本身的Consumer Group管理机制。所以不同Flink job 即使使用同样的 group id 也会消费到同样的消息。 On Mon, Apr 22, 2019 at 1:24 PM 13341000780 <13341000...@163.com> wrote: > hi, 各位大牛好！ > kafka作为数据源时，在 Properties > 中设置了group.id，信息，但是同一个group下的consumer中能消费到同一个主题下想同一条消息。 > 另外我单独使用 kafka consumer时就

flink-kafka 消费组问题

hi, 各位大牛好！ kafka作为数据源时，在 Properties 中设置了group.id，信息，但是同一个group下的consumer中能消费到同一个主题下想同一条消息。 另外我单独使用 kafka consumer时就不会出现这种情况，即同一个消费组下只能有一个 consume 消费到同一条消息。 各位大牛有人遇到过这个问题吗？望赐教。 非常感谢.