Kafka Connect Bigquery接收器-关闭时偏移量提交失败
Kafka Connect BigQuery接收器是一种用于将Kafka消息流式传输到Google BigQuery的工具。它允许将Kafka中的数据实时导入到BigQuery中，以便进行进一步的分析和处理。
关闭时偏移量提交失败是指在关闭Kafka Connect BigQuery接收器时，无法成功提交偏移量。偏移量是用于跟踪Kafka消息的位置的标识，以确保数据的准确性和一致性。当偏移量提交失败时，可能会导致数据丢失或重复处理。
解决关闭时偏移量提交失败的问题，可以采取以下步骤：
- 检查网络连接：确保Kafka Connect BigQuery接收器与Kafka集群和BigQuery服务之间的网络连接正常。检查网络配置、防火墙设置等，确保没有阻止或限制连接的问题。
- 检查权限设置：确保Kafka Connect BigQuery接收器具有足够的权限来提交偏移量。检查相关的身份验证和授权配置，确保接收器具有正确的访问权限。
- 检查配置参数：检查Kafka Connect BigQuery接收器的配置参数，确保正确设置了偏移量提交的相关参数。例如，可以检查偏移量提交的频率、提交超时时间等参数，确保其适合当前的环境和需求。
- 日志和错误排查：查看Kafka Connect BigQuery接收器的日志和错误信息，以了解具体的失败原因。根据错误信息进行排查和调试，可能需要查看相关的文档或寻求技术支持。
推荐的腾讯云相关产品和产品介绍链接地址：
腾讯云提供了一系列与云计算相关的产品和服务，包括云服务器、云数据库、云存储等。具体针对Kafka Connect BigQuery接收器的解决方案，可以参考腾讯云的消息队列 CMQ（Cloud Message Queue）和数据仓库 TDSQL（TencentDB for TDSQL）等产品。
CMQ产品介绍链接：https://cloud.tencent.com/product/cmq TDSQL产品介绍链接：https://cloud.tencent.com/product/tdsql
请注意，以上推荐的腾讯云产品仅供参考，具体的选择应根据实际需求和情况进行评估和决策。
相关·内容
1回答
分布式模式下Kafka Connect集群的Offset Management的行为是什么，即运行多个Connector并监听同一组主题(或一个主题)？ 因此，在分布式模式下，Kafka Connect将偏移量信息存储在Kafka中，该偏移量将被集群中的workers读取并提交。如果在该Kafka Connect集群中有多个Connector在监听同一主题，会发生什么情况？分区的偏移量是所有连接器的偏移量都相同，还是每个连接器在分区上的偏移量不同？
浏览 15提问于2019-07-01得票数 2
我正在使用properties/json文件配置连接器，在从源连接器读取消息时，我尝试添加一个时间戳列，其中包含kafka时间戳，但没有成功。
我尝试过添加transforms，但是它总是null，我的接收器连接器“大查询”返回了一个错误
更新表架构失败
我把这些配置放在bigquery连接器属性中
transforms=InsertField
transforms.InsertField.timestamp.field=fieldtime
transforms.InsertField.type=org.apache.kafka.connect.transforms.InsertFie
浏览 0提问于2018-11-15得票数 1
回答已采纳
我尝试使用Confluent BigQuerySinkConnector创建Kafka到BigQuery的数据管道。测试环境是一个 docker容器。我在上面添加了(它不是默认存在的)。我在谷歌BigQuery端做的所有定义(我希望...)。但它只是给出了Schema Registry错误，我不能理解为什么会发生这种情况。我在BigQuery dataset上创建了一个名为rest_avro的表。主题模式：
{
"fields": [
{
"name": "name",
"type": "
浏览 0提问于2020-08-19得票数 1
1回答
我的flink作业使用FlinkKafkaConsumer010从kafka consumer读取数据，并使用CustomBucketingSink接收到hdfs。我们有一系列的转换kafka -> flatmaps(2-3个转换) -> keyBy -> tumblingWindow(5 mins) -> Aggregation -> hdfsSink。我们有平均300万/分钟事件的kafka输入，高峰时间大约2000万/分钟的事件。检查点持续时间和两个检查点之间的最小停顿是3分钟，我使用的是FsStateBackend。
以下是我的假设：
Flink从kafk
浏览 0提问于2020-03-12得票数 0
我使用hdfs连接集群，就像在分布式模式下一样。
我设定rotate.interval.ms为1小时，offset.flush.interval.ms为1分钟。
在我的例子中，我认为当一个带有第一个记录时间戳的时间间隔为一个小时的新记录出现时，文件就会被提交；偏移量将每分钟刷新一次。
但是，当文件仍然打开时，我想知道重新启动集群时会发生什么。我是说，在下面的情况下会发生什么？
该文件是以带有“15:37”时间戳的记录开始打开的。(抵销10)
10分钟后，kafka连接集群重新启动。
(我认为步骤1的文件将被丢弃在内存中，而不是提交给hdfs)
当新员工开始工作时，“新打开的文件
浏览 3提问于2021-12-20得票数 0
回答已采纳
我有一个来自网页的流数据加载到BigQuery中，我想将BigQuery连接到卡夫卡作为源。有没有像kafka-connect这样的连接器用于BigQuery接收器。
谢谢，巴拉
浏览 1提问于2018-02-20得票数 3
我想重新发布x次处理失败的消息。我看到了手动提交代码：
val consumerWithOffsetSink = kafka.consumeWithOffsetSink(consumerProperties)
Source.fromPublisher(consumerWithOffsetSink.publisher)
.map(processMessage(_)) // your message processing
.to(consumerWithOffsetSink.offsetCommitSink) // stream back for commit
.run()
但是，如何处理proc
浏览 0提问于2017-12-16得票数 2
我使用的是kafka jdbc接收器连接器，它从主题插入数据到SQL Server数据库。由于mssql中的某些问题，连接器已停止，并出现错误：
Caused by: org.apache.kafka.connect.errors.ConnectException: com.microsoft.sqlserver.jdbc.SQLServerException: The TCP/IP connection to the host mssql_ip , port 1433 has failed. Error: "connect timed out.
我的问题是，一旦SQL Server启
浏览 4提问于2020-04-03得票数 1
我使用Kafka从Kafka Broker (v0.10.2)获得消息，然后将其同步到下游服务。
目前，我在SinkTask#put中有处理SinkRecord的代码，然后将其持久化到下游服务。
几个关键要求，
我们需要确保消息至少保存到下游服务一次。
如果下游服务抛出一个错误或者说它没有处理消息，那么我们需要确保消息再次被重新读取。
因此，我们认为我们可以依赖SinkTask#flush，通过抛出异常或告诉Connect不要提交偏移，但在下一次投票中重试，从而有效地退出对特定轮询/接收消息周期的提交偏移。
但正如我们所发现的，flush实际上是基于时间的&或多或少独立
浏览 2提问于2017-07-19得票数 9
我正在设计一个apache storm拓扑(使用streamparse)，它使用一个spout (apache kafka spout)和一个并行大于1的螺栓构建，可以批量读取kafka spout中的消息并将消息持久化到mysql表中
螺栓分批读取消息。如果批处理成功完成，我将手动提交apache kafka偏移量。
当螺栓在mysql上插入失败时，我不会在kafka中提交偏移量，但一些消息已经在spout发送到螺栓的消息队列中。
已经在队列中的消息应该被删除，因为我不能在不丢失先前失败的消息的情况下推进kafka偏移量。
streamparse中有没有一种方法来清理或失败所有在螺栓启动时已
浏览 2提问于2019-01-31得票数 0
1回答
我试图在Java上实现Kafka消费者。
假设使用者包含一些可能引发异常的消息处理逻辑。在这种情况下，使用者应该睡上一段时间，并重新处理最后一条消息。
我的想法是使用手动偏移管理:偏移量不会在失败时提交，因此使用者可能会从旧的偏移量中读取。
在测试期间，我发现消息实际上只读取一次，尽管没有提交偏移量。上次提交的偏移量仅在应用程序重新启动时考虑。
我的问题是：
我做得对不对？
人工偏移管理的用法是什么？
浏览 2提问于2016-09-19得票数 0
回答已采纳
我正在努力建立一个卡夫卡连接分布式模式应用程序，这将是一个卡夫卡到S3管道。我正在使用Kafka 0.10.1.0-1和Kafka Connect 3.1.1-1.到目前为止，事情进展顺利，但是对于我正在处理的更大的系统来说，一个很重要的方面需要了解卡夫卡-> FileSystem管道的偏移信息。根据文档，offset.storage.topic配置将是分布式模式应用程序用于存储偏移信息的位置。考虑到卡夫卡如何在“新”卡夫卡中存储消费者补偿，这是有意义的。但是，在对FileStreamSinkConnector进行了一些测试之后，没有任何东西被写入到我的offset.storage.to
浏览 9提问于2017-01-05得票数 2
1. env
卡夫卡版本: 0.11.0.1。
2. Problem desc
无论多少个kafka节点，如果停止头两个kafka过程，消费者就无法从kafka获得消息(生产者可以向kafka发送消息)，经过检查，头两个kafka节点是消费者组协调器，如果一个挂起，另一个继续工作，如果挂两个，消费者都不工作，报告错误:偏移提交失败，这不是正确的协调者。
3.我的问题.
我的测试结果正确吗？如果测试结果是正确的，卡夫卡似乎不支持高可用。
如果我的测试结果是错的，有什么问题吗？
浏览 0提问于2018-06-07得票数 0
回答已采纳
在不提供模式的情况下保存到bigquery中有一个问题。
Kafka连接Config
tasks.max: 1
topics: sample_topic
project: gcp-project-name
defaultDataset: dataset-name
keyfile: key_file_path
group.id: bq-connector
confluent.license: licence_file_path
错误
[2021-12-01 11:10:16,554] ERROR Task failed with com.wepay.kafka.connect.bigquery.
浏览 13提问于2021-11-30得票数 0
我了解到Kafka connect可以在集群模式下部署。并且工作人员在数据源和kafka主题之间移动数据。我想知道的是，如果工作人员在数据源之间移动数据到kafka主题时失败，会不会有数据丢失？如果发生数据丢失，我们如何从连接器获取数据，或者kafka connect会自动处理它吗？
浏览 33提问于2021-01-22得票数 0
回答已采纳
我已经为我的数据存储(Amppol )开发了kafka-sink-connector (使用confluent-oss-3.2.0-2.11，connect framework)，它将kafka主题中的数据存储到我的存储中的相应表中。
只要kafka服务器和ADS服务器启动并运行，一切都像预期的那样工作。
需要关于特定用例的帮助/建议，其中事件在kafka主题中被摄取，而其下的接收器组件(ADS)已关闭。这里的期望是，每当接收器服务器出现时，先前在kafka主题中摄取的记录应该插入到表中；
请指教如何处理这样的情况。
connect框架中是否有对此的支持？或者至少一些参考资料会有很大的帮助。
浏览 1提问于2017-04-04得票数 0
1回答
我对kafka API有以下问题。我用以下命令设置我的客户：
<code>A0</code>
然后
<code>A1</code>
我想从Kafka中读取数据，从这些数据中更新数据库。但如果更新失败，我想重试，直到它工作。所以我想将DB事务应用到kafka，i.a。如果我的数据库事务是OK的，那么移动kafka指针，但如果它失败了，那么从相同的位置重试。
在我的代码中，
<code>A2</code>
没有像预期的那样工作，这意味着“如果kafka崩溃，那么从提交的位置重新启动”。但是，当我的DB事务失败时，即使我不使用c
浏览 31提问于2018-12-21得票数 1
回答已采纳
如何获取、设置或重置Kafka Connect连接器/任务/接收器的偏移量？
我可以使用运行kafka.admin.ConsumerGroupCommand的/usr/bin/kafka-consumer-groups工具来查看我所有常规Kafka消费者组的偏移量。但是，Kafka Connect任务和组不会显示在此工具中。
类似地，我可以使用zookeeper-shell连接到Zookeeper，我可以看到常规Kafka消费者组的zookeeper条目，但不能看到Kafka connect接收器的zookeeper条目。
浏览 1提问于2016-06-09得票数 16
回答已采纳
1回答
我正在为Kafka connect编写HTTP Sink插件。它的作用是为配置的Kafka主题中的每条消息发送HTTP请求。我希望将消息发送到死信队列，以防HTTP请求失败。我可以使用接收器插件中提供的死信队列配置吗？ 这个问题的原因是，在kafka connect文档和一些博客中已经提到，只有转换器和转换器中的错误才会被发送到死信队列，而不是PUT过程中的错误。因为发送http请求的任务是在PUT中完成的。所以我在想，有没有办法向DLQ发送失败的http消息？如果没有，是否可以将消息发送到其他kafka主题进行进一步处理？
浏览 26提问于2020-08-31得票数 1
3回答
我有一个Java Akka应用程序，从Kafka读取，处理消息，并手动提交。
我使用的是0.10.1.1 API的高级使用者。
奇怪的是，当我关闭应用程序并重新启动它时，偏移量比上次提交时要大一点，我找不到原因。
我在代码中只有一个提交点。
else if(message.getClass() == ProcessedBatches.class) {
try {
Logger.getRootLogger().info("[" + this.name + "/Reader] Commiting ...");
浏览 1提问于2017-05-25得票数 1
我创建了一个简单的Apache光束流管道，它从Kafka读取数据，进行一些处理，并通过调用一些外部服务的API来持久化结果。我希望确保在管道重启或失败期间不会丢失数据，因此我希望在特定doFun执行结束时成功调用接口后，手动将记录偏移量提交给Kafka。 在我之前的Kafka经验中，我知道通过使用Kafka Consumer的below接口，我可以手动将记录偏移量提交给Kafka。 consumer.commitSync(currentOffsets); 在KafkaIO设置中有关闭自动提交的设置，但是我没有找到任何有效的解决方案来手动提交偏移量，因为似乎没有办法在doFun中访问消费者。
浏览 10提问于2021-09-21得票数 1
回答已采纳
我正在尝试设置Kafka Connect BigQuery sink connector。我有大约50个kafka主题。当我部署连接器时，最初的10(我认为这是连接器配置中的threadPoolSize默认值)表显示了一些数据。然后，新数据将不再传入表中。此外，BigQuery中也不会显示新的表。 我的连接器配置： {
"name": "kcbq-connect1",
"config": {
"connector.class": "com.wepay.kafka.connect.bigquery.B
浏览 9提问于2020-12-05得票数 0
我是指Flink 1.14版本的Kafka源代码连接器下面的代码。
我期待以下要求。
在应用程序非常新的开始时，必须从检查点上卡夫卡主题的最新偏移量中读取，它必须在重启后(当应用程序手动/系统错误终止时)将所消耗的偏移量提交给Kafka ，它必须从上次提交的偏移量中选择，并且必须使用消费者延迟，从此以后还要使用新的事件提要。
使用新的KafkaConsumer API (KafkaSource)，我面临以下问题
能够执行上述要求，但不能在检查点(500 on )上提交所消耗的偏移量。而是在2s或3s.之后提交。
当您在2s/3s内手动关闭应用程序并重新启动时。由于上次使用的消息未提交，因此将读
浏览 10提问于2021-12-16得票数 2
我想查询Kafka主题中的消息，但不是所有消息，不是从头开始。我只需要查看哪些消息尚未根据消费者组提交。所以，基本上我想要的是删除偏移量低于消费者组偏移量的单据。
在这一点上，如果我使用elastic-connector，有没有办法或变通办法在消息被使用和提交后从弹性索引中删除文档？
或者，我应该使用Kafka Streams以及如何使用？
浏览 2提问于2021-05-08得票数 0
我正在考虑使用Kafka连接器和创建我自己的Kafka消费者/生产者来将一些数据从Kafka转移到Kafka，我看到了Kafka连接器在可扩展性和容错方面所提供的价值。但是，如果“任务”由于某种原因而失败，我无法找到连接器的确切行为。以下是几种场景：
对于接收器连接器(S3-Sink)，如果它(任务)成功地将数据发送到目标(例如由于网络问题)失败(在所有重试之后)，那么工作人员会发生什么情况？它会坠毁吗？它能够再消费同样的数据从Kafak以后吗？
对于源连接器(JDBC )，如果它不能发送给Kafka，它以后会重新处理相同的数据吗？这取决于来源是什么？
以上问题的答案是否取决于
浏览 0提问于2018-08-22得票数 1
我使用过Debezium作为Mysql的-> Elasticsearch。现在的问题是，当我从MySQL中删除数据时，它仍然会在Elasticsearch中重新出现，即使数据不再存在于MySQL DB中。UPDATE和INSERT工作正常，但DELETE不是。另外，我做了以下工作：
Delete在MySQL中的数据
Delete Elasticsearch索引和ES Kafka Sink
Create -- Kakfa中ES的新连接器
现在，奇怪的是，所有我删除的数据收割机在这里，以及！当我在步骤(3)之前检查ES数据时，数据不存在。但之后，这种行为会被观察到。
请帮我解决这个问题！
M
浏览 7提问于2021-03-14得票数 0
我正在写一个使用投票机制的kafka消费者，在这个机制中，我每次投票都会收到100条消息。在消费消息之后，我一个接一个地手动提交偏移量。在提交偏移量时，有时我会收到100条消息中的一条消息的错误。Rest偏移量提交成功。 Offsets
1,2,3,4,5,6,7,....100 假设偏移量5提交失败，但不包括偏移量5，所有偏移量都成功提交。 那么，在这种情况下，提交失败的偏移量会发生什么？由于偏移量是按顺序移动的，我是否会在下一次轮询中获得提交失败的偏移量？ 我的初步调查显示，它在下一次轮询中读取失败的偏移量。需要在这方面的专业意见。
浏览 15提问于2019-04-18得票数 1
回答已采纳
有没有可能使用Kafka源代码，在Beam中使用KafkaIO进行一次处理？有一个可以设置的名为AUTO_COMMIT的标志，但它似乎是在使用者处理数据之后直接提交给Kafka，而不是在管道完成处理消息之后。
浏览 2提问于2018-01-18得票数 3
回答已采纳
查看我对解决方案的公认答案的最后评论
我配置了一个DStream，如下所示：
val kafkaParams = Map[String, Object](
"bootstrap.servers" -> "kafka1.example.com:9092",
"key.deserializer" -> classOf[StringDeserializer],
"value.deserializer" -> classOf[KafkaAvroDeserializer],
"g
浏览 3提问于2018-07-04得票数 0
回答已采纳
如果我实现Kafka Connect与Source and Sink连接器同步2个表，是否有可能在以后更改Kafka平台而不会对表数据产生任何影响。
我使用Kafka连接在微服务数据库之间同步数据，我预测会有一些实例需要从一个Kafka实例移动到另一个实例。
场景:如果我在表A中有100条记录，那么在表B中也会有同样的100条记录可用，并且使用主题T1进行同步。如果我丢弃现有的Kafka topic T1并为同一任务创建一个新的topic T2 (或者如果我从一个Kafka集群移到另一个集群)，是否会再次将表A中的100条记录带到表B中，从而导致重复？
浏览 0提问于2020-03-11得票数 1
我需要使用kafka连接接收器将文件数据发送到ftp服务器，在服务器接收到整个文件后，我还必须发送确认来更改数据库中这些任务的状态。
我想知道在这里最好的方法是什么，我最初想创建一个自定义的ftp kafka connect，它也会改变任务数据库的状态。这是最好的方法吗，还是有其他选择？
浏览 0提问于2017-05-30得票数 0
我正在尝试实现简单的服务，从kafka中提取消息，将它们包装在一些数据中，然后发送到外部服务。
在处理消息时，处理外部服务不可用的常见模式是什么？
到目前为止，只有当对外部服务的请求成功时，我才手动提交消息。我希望kafka在一段时间后重新发送消息，如果消息没有被提交，那么处理外部服务故障对消费者来说是透明的。不过，我找不到这样做的方法。然而，我很好奇，如果我没有做一些反模式，并且有更好的解决方案。
浏览 4提问于2016-10-01得票数 0
我在Kubernetes上运行Kafka Connect (8-16个节点，自动标度)。我总共定义了44个连接器，每个Kafka主题一个(每个主题一个分区)。这些主题是由Debezium / Postgresql编写的。有3个Kafka节点。每个连接器的tasks.max设置为4。大部分连接器(但不是每个连接器！)有一个(总是一个)失败的任务，由于java.lang.IllegalStateException:没有当前分配的分区-0。
这里不是卡夫卡专家，请注意;)我假设有3个Kafka节点，因此3个工作人员做得很好，而第四个任务没有任何连接，所以失败了。但是为什么有时有4个任务运行得很好呢？
浏览 3提问于2019-03-24得票数 0
回答已采纳
我有一个Streaming应用程序，它从Kafka主题中读取数据并将其放在HDFS上。我希望应用程序存储抵消读到__consumer_offsets主题的消息，以便在应用程序失败时开始阅读。这个应用程序运行良好(我可以看到HDFS上的数据)，但我看不到它对__consumer_offsets的提交。
这是我的KafkaParams：
val kafkaParams = Map(
"metadata.broker.list" -> "xx.xxx.x.xx:6667",
"enable.auto.commit" -&g
浏览 0提问于2018-11-08得票数 0
我使用的是集成了streaming-kafka的流媒体。
我的kafka主题有80个分区，而我的机器有40个核心。我发现当作业运行时，kafka消费者进程仅部署到2台机器(40*2=80)上，这2台机器的带宽将非常非常高。
我想知道有没有办法控制kafka消费者的调度，以平衡带宽和内存使用？
浏览 1提问于2015-05-13得票数 1
尝试使用以下配置创建elasticsearch接收器连接器，创建成功，但当在"my.topic.one“上生成消息时，ES接收器连接器在尝试创建名为"my.topic.one”的索引时失败：“无法创建索引'my.topic.one'”(我用来连接ES的用户故意没有create index权限)。为什么它试图创建一个新的索引，以及如何让连接器索引到以前创建的"elasticsearch_index_name"？
{
"type.name": "_doc",
"
浏览 19提问于2020-05-20得票数 0
1回答
我需要将消息接收到我的BigQuery表，我想知道使用这些消息的最佳方式是什么。我在AWS的Kafka服务器会生成AVRO消息，根据我所看到的，Dataflow需要接收JSON格式的消息。因此，我搜索并找到了一篇文章，解释了如何向PubSub接收消息，但在PubSub上，我只看到在这种体系结构中，他们在GCP上创建了一个Kafka来生成消息。我需要知道的是：
可以从外部的Kafka服务器在PubSub上接收AVRO消息，然后使用messages反序列化消息，将其发送到Dataflow，并最终将其发送到BigQuery表？
还是我需要创建一个Kafka并使用它来使用来自外部服务器
浏览 1提问于2021-04-27得票数 3
回答已采纳
1回答
我的问题分成两部分。我读过。我想要完全删除一个连接器，带有偏移和所有，这样我可以重新创建它的同名以后。这个是可能的吗？据我所知，墓碑信息将无限期地扼杀这个连接器。
第二部分是--是否有办法让kafka-connect容器在关闭时自动删除他创建的所有连接器？谢谢
浏览 7提问于2022-11-17得票数 1
无模式的Bigquery接收器连接器SMT无法在布尔值上将数据保存到bigquery。
MapsUtil.debugPrint on recordValue，然后从apply(R record)返回。
active = true java.lang.String
模式定义
{
"mode": "NULLABLE",
"name": "active",
"type": "BOOLEAN"
}
去列剂
public class BooleanDeserialiser ex
浏览 15提问于2022-01-25得票数 1
回答已采纳
1回答
我在centos 7(汇合)上安装了Apache，我试图在分布式模式下运行filestream连接，但是我得到了以下错误：
[2017-08-10 05:26:27,355] INFO Added alias 'ValueToKey' to plugin 'org.apache.kafka.connect.transforms.ValueToKey' (org.apache.kafka.connect.runtime.isolation.DelegatingClassLoader:290)
Exception in thread "main" o
浏览 6提问于2017-08-10得票数 1
回答已采纳
我试图从一个卡夫卡主题流到谷歌BigQuery。我的connect-standalone.properties文件如下：
bootstrap.servers=rhes75:9092,rhes75:9093,rhes75:9094,rhes564:9092,rhes564:9093,rhes564:9094,rhes76:9092,rhes76:9093,rhes76:9094
key.converter=org.apache.kafka.connect.storage.StringConverter
##value.converter=org.apache.kafka.connect.stor
浏览 1提问于2021-03-13得票数 0
回答已采纳
我正在尝试使用s创建一个卡夫卡接收器连接器。但是，出于某种原因，日志输出正在报告一个SourceConnectorConfig：
INFO ConnectorConfig values:
connector.class = com.spredfast.kafka.connect.s3.sink.S3SinkConnector
key.converter = null
name = transactions-s3-sink
tasks.max = 1
transforms = null
value.c
浏览 0提问于2017-10-21得票数 0
回答已采纳
1回答
我用java产生数据，然后把它存入Kafka topic，然后我想把这些数据存入MongoDB。当我通过JAVA以JSON的形式发送数据时，由于这个错误，它不会存储到MongoDB中。
[2020-08-15 18:42:19,164] ERROR WorkerSinkTask{id=Kafka_ops-0} Task threw an uncaught and unrecoverable exception. Task is being killed and will not recover until manually restarted. Error: JSON reader was e
浏览 2提问于2020-08-16得票数 0
我正在编写一个测试flink两步提交的案例，下面是概述。
sink kafka正是卡夫卡的制作人。sink step是mysql接收器扩展two step commit。sink compare是mysql接收器扩展two step commit，这个接收器偶尔会抛出一个exeption来模拟检查点失败。
当检查点失败并恢复时，我发现mysql两步提交将运行良好，但是kafka消费者将读取与上一次成功的抵消，甚至在这个检查点失败的之前他也会生成消息。
在这种情况下，如何避免重复消息？
谢谢你帮忙。
env
flink 1.9.1
java 1.8
kafka 2.11
浏览 6提问于2019-12-05得票数 1
回答已采纳
从Kafka connect中删除源连接器后，为该连接器保存的偏移量仍然存在于Kafka中。当具有相同名称的相同连接器再次启动时，它将从上次保存的偏移量开始处理数据。更改连接器的名称不是一个解决方案，因为它使得连接器提交很难自动化。 是否有解决此问题的方法(重命名源连接器不是一个选项)
浏览 77提问于2021-06-17得票数 2
回答已采纳
我们正试图通过与启用IAM的MSK连接。
下面的属性被添加到$KAFKA_HOME/config/connect-standalone.properties中，除了需要的常规信任之外。
需要sasl.jaas.config=software.amazon.msk.auth.iam.IAMLoginModule security.protocol=SASL_SSL sasl.mechanism=AWS_MSK_IAM；sasl.client.callback.handler.class=software.amazon.msk.auth.iam.IAMClientCallbackHandler
浏览 10提问于2022-08-24得票数 0
回答已采纳
在这种情况下，我如何回滚/撤消偏移量提交(参见catch块中的注释)？ public void HandleMessage(ConsumeResult<object, int> messageResult, IConsumer consumer) {
try {
consumer.commit(messageResult);
dbProvider.startTransaction();
//Do db operations
dbProvider.commit();
}
浏览 9提问于2021-04-04得票数 0
回答已采纳
1回答
在第一次创建事件接收器类型kafka是成功的。但是，当我再次运行kafka并从wso2 cep创建一个新的事件接收器时。
它将如下所示：
添加事件接收器失败，例外: java.lang.NoClassDefFoundError: org/apache/kafka/Failed/security/JaasUtils
添加新的事件接收器失败。
我该怎么处理那个案子？
浏览 3提问于2016-05-16得票数 0
高级消费者API似乎是一次读取一条消息。
如果消费者想要处理这些消息并将其提交给其他下游消费者，比如Solr或Elastic-Search，这对他们来说可能是一个很大的问题，因为他们更喜欢批量发送消息，而不是一次只有一个。
在内存中对这些消息进行批处理也不是一件容易的事，因为Kafka中的偏移量只有在批处理已经提交时才需要同步，否则具有未提交下游消息的崩溃的kafka消费者(如Solr或ES)的偏移量将已经更新，从而导致消息松散。
如果消费者在向下游提交消息之后但在更新消息偏移量之前崩溃，那么它可能会多次使用消息。
如果Kafka批量使用消息，那么将非常感谢一些指向代码/文档的指针。
谢谢!
浏览 2提问于2016-03-04得票数 9
我目前正在编写一个Spark流媒体应用程序，它从Kafka读取数据，并在应用一些转换之前尝试解码它。
当前的代码结构如下所示：
val stream = KafkaUtils.createDirectStream[String, String](...)
.map(record => decode(record.value())
.filter(...)
.foreachRDD { rdd =>
val offsetRanges = rdd.asInstanceOf[HasOffsetRanges].offsetRanges
...
stream.asIns
浏览 0提问于2017-10-31得票数 5
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
即时通信 IM
实时音视频活动推荐
腾讯云开发者
