为什么Kafka ProducerListener没有记录跟踪id和span id？
Kafka ProducerListener没有记录跟踪id和span id的原因是因为Kafka本身是一个分布式消息队列系统，它的设计目标是高吞吐量和低延迟，而不是为了实现分布式追踪和跟踪功能。
跟踪id和span id通常用于分布式追踪系统，如Zipkin、Jaeger等，用于跟踪和监控分布式系统中的请求流程和性能。这些系统通常通过在请求中添加唯一的跟踪id和span id，并在请求经过不同的服务时进行传递和记录，从而实现对请求的全链路追踪和性能分析。
然而，Kafka并没有内置的分布式追踪功能，因此在Kafka的ProducerListener中没有记录跟踪id和span id的机制。如果需要在Kafka中实现分布式追踪，可以考虑以下两种方式：
- 在消息的header中添加跟踪id和span id：可以在发送消息时，在消息的header中添加唯一的跟踪id和span id，并在ProducerListener中获取并记录这些信息。这样可以在消息经过不同的服务时，将这些信息传递下去，但需要注意的是，这种方式需要在应用程序中手动实现。
- 结合分布式追踪系统：可以将Kafka与分布式追踪系统集成，如通过使用Kafka的拦截器（interceptor）来实现。拦截器可以在消息发送和接收的过程中进行处理，可以在拦截器中获取跟踪id和span id，并将其传递给分布式追踪系统进行记录和分析。
需要注意的是，以上两种方式都需要在应用程序中进行额外的开发和配置，并且需要与分布式追踪系统进行集成。具体的实现方式和配置方法可以根据具体的需求和使用的分布式追踪系统来确定。
腾讯云相关产品和产品介绍链接地址：
- 腾讯云消息队列 CKafka：https://cloud.tencent.com/product/ckafka
- 腾讯云分布式追踪系统 TSW：https://cloud.tencent.com/product/tsw
相关·内容
我有一个Kafka实例和一个简单的spring引导应用程序，其中包含一个REST控制器和一个ProducerListener bean。控制器接受一条简单的String消息并通过KafkaTemplate将其发送给Kafka。我希望ProducerListener记录一条info消息，但是缺少跟踪id和span id。在我看来，它应该包括在日志中。 我使用Spring Cloud Sleuth和Spring Kafka starters。 消息本身通过Kafka主题成功地发送到另一个spring启动应用程序，该应用程序正确地获取了跟踪id，所以我假设问题只与ProducerListener有
浏览 27提问于2019-10-04得票数 1
1回答
我是第一次在scala中使用kafka。我试图消费卡夫卡主题的数据，该主题已经由制片人发送。我的代码如下：
consumer.poll(5000).asScala.toList
我的卡夫卡主题已经有49张唱片了。当我用这个的时候，它带来了随机记录。当我打印偏移量和时间戳时，它开始打印从0到21的偏移量。但当我检查卡夫卡主题0偏移时间戳是不同的。它们都不是最后的21项记录。没有检查所有的记录，但检查了一些从最后和顶部。意味着它没有选择实际的0偏移量数据。它为什么这样做？
我的消费者吐露如下：
val props = new Properties()
props.put("
浏览 4提问于2020-07-28得票数 0
2回答
我需要向集群中运行的Flink作业添加track和span，请求流如下所示
用户-> Rest ->卡夫卡-主题1-> FlinkJob-1 ->Kafka-主题2-> FlinkJob-2 -> Consumer -> DB
我使用Spring引导来创建rest API，并使用spring Sleuth向生成的日志添加跟踪和span id，在调用rest和将消息放到Kakfa-Top-1上时添加跟踪和span id，但我无法弄清楚如何在using作业-1和FLinkJob 2中使用消息时添加跟踪和span id，因为它们脱离了春季上下文。
一种方法是
浏览 3提问于2019-11-22得票数 0
假设我们正在KStream和KTable之间进行内部连接，如下所示：
StreamsBuilder sb = new StreamsBuilder();
JsonSerde<SensorMetaData> sensorMetaDataJsonSerde = new JsonSerde<>(SensorMetaData.class);
KTable<String, String> kTable = sb.stream("sensorMetadata",
Consum
浏览 2提问于2020-06-06得票数 0
回答已采纳
我能够开发一个从kafka读取的管道，做一些转换，并将输出写入kafka接收器以及parque接收器。我希望添加有效的日志记录来记录转换的中间结果，就像在常规的流应用程序中一样。
我看到的一种选择是通过以下方式记录queryExecutionstreams
df.queryExecution.analyzed.numberedTreeString
或
logger.info("Query progress"+ query.lastProgress)
logger.info("Query status"+ query.status)
但这似乎没有办法查看流在其上
浏览 13提问于2018-09-05得票数 3
1回答
我想使用Kafka流，但也与外部数据库和API交互。我读到过有人建议使用Kafka Connect这样做，但我想知道怎么做和为什么。即使没有Kafka流，也可以查询数据库(即使它没有Kafka连接连接器)和每个记录的API吗？或者我应该为这个用例使用另一个处理引擎？
浏览 3提问于2022-01-15得票数 0
1回答
我使用Kafka实现微服务之间的Asyng调用，使用Spring进行日志记录。日志记录是正常的，但是当有消息从Microservice1到Microservice2时，日志记录的消息具有不同的跟踪ID。难道他们不需要有相同的追踪Id但有不同的SpanId吗？有没有特殊的配置？
浏览 2提问于2016-12-02得票数 4
回答已采纳
我想在从producer发布数据时查看要在记录中引用的模式的Id/版本。 如果我有3个版本的模式，并且我正在发布版本1的数据，我可以在将记录发送给kafka之前查看版本号吗？
浏览 10提问于2020-01-02得票数 0
回答已采纳
我正在设计一个数据管道，它从读取的平面文件开始。文件中的每一行都是一条记录。
加载后，将对每条记录进行解析、转换和丰富。这是独立于其他记录发生的。
作为最后一步，我想要基于几个记录字段的模糊匹配来删除记录。要做到这一点，我想得到2个记录的所有组合。
目前我使用sql表作为缓冲区。我的表包含所有记录，我将表与自身连接，使用sounds like模糊匹配键不同的条件，并对名称进行模糊匹配
CREATE TABLE temp_tblSoundsLikeName AS
SELECT DISTINCT clients1.client_name client_name1,
浏览 1提问于2017-09-27得票数 0
我使用HDFS接收器连接器，我想知道有多少记录被放入HDFS。
在日志中，当连接器开始将文件放入HDFS时，我有日志记录，但没有记录多少记录。
例如：
INFO Opening record writer for: hdfs://hdfs/path/+tmp/table/partition=2020-02-27/19955b52-8189-4f70-94b5-46d579cd1505_tmp.avro (io.confluent.connect.hdfs.avro.AvroRecordWriterProvider)
通过扩展连接器本身是可能的吗？
我使用kafka连接HDFS 2接收器。
浏览 3提问于2020-03-02得票数 0
回答已采纳
我需要知道是否可以为Apache Kafka broker配置日志记录，以编写所有产生/消费的主题和消息。我一直在寻找log4j.properties，但建议的属性似乎都不能满足我的需要。
提前谢谢。
查看Kafka生成的日志文件，它们似乎都没有记录在不同主题中编写的消息。
更新:不完全是我想要的，但对于任何看起来相似的人，我找到了：，它提供了一个友好的图形用户界面来查看不同主题的消息。
浏览 0提问于2019-06-25得票数 1
我使用的是Kakfa https://www.confluent.io/hub/mongodb/kafka-connect-mongodb源代码连接器，它使用的是Connector v5.4。下面是我的MongoDB源连接器配置
{
"name": "mongodb-replica-set-connector",
"config": {
"tasks.max": 1,
"connector.class": "com.mongodb.kafka.connect.
浏览 0提问于2020-07-18得票数 2
回答已采纳
我是Kafka Java API的新手，我正在使用来自特定Kafka主题的记录。
我知道我可以使用subscribe()方法从主题开始轮询记录。如果我想要从选定的主题分区开始轮询记录，Kafka还提供了assign()方法。
我想知道这是不是两者唯一的区别？
浏览 1提问于2018-12-27得票数 17
回答已采纳
1回答
目标:读取主题中的所有消息，然后终止该过程。
我可以通过以下内容连续阅读消息：
props.put("bootstrap.servers", kafkaBootstrapSrv);
props.put("group.id", group_id);
props.put("max.poll.records", 1); // Only get one record at a time. I understand that to read all messages this will need to be increased
props.put("
浏览 16提问于2018-12-14得票数 3
回答已采纳
我正在使用Debezium SQL Server Source connector通过CDC将数据推送到Kafka。SQL Server表中的数据是连续插入的。当Kafka宕机时，数据还会插入到该表中吗？如果是这样，当我的Kafka打开时，如何获得这些记录？
浏览 13提问于2020-05-22得票数 0
回答已采纳
我设置了一个Flink工作，向Kafka主题写一些消息。它使用FlinkKafakProducer作为接收器。我验证了输出Kafka主题中写入了消息，但Flink指标numRecordsOut始终显示0。同样，在Flink Web UI中，发送的字节也始终显示0B。
记录输出数量不能反映实际数量的原因是什么？
浏览 20提问于2019-03-28得票数 1
回答已采纳
有没有一种优雅的方式来查询特定记录的Kafka主题？我正在构建的REST API获得一个ID，并且需要在Kafka主题中查找与该ID相关联的记录。一种方法是通过自定义消费者检查主题中的每条记录，并查找匹配的记录，但我希望避免读取大量记录的开销。Kafka有没有快速内置的过滤功能？
浏览 3提问于2016-03-01得票数 19
回答已采纳
我正在尝试使用Jdbc接收器连接器将数据从Kafka主题加载到Postgres。现在，我们如何知道到目前为止加载到Postgres中的记录的数量。到目前为止，我一直在使用sql查询检查数据库中的记录数量。有没有其他方法可以让我知道呢？
浏览 32提问于2021-06-17得票数 1
1回答
风暴喷口直接从MySQL？
、、、、
我想使用Storm集群来实时监控用户活动。目前，用户活动日志保存在MySQL中。每分钟大约有100,000条记录被记录到MySQL中，每天有20,000,000条记录。我想做一个topN查看项目排名每N分钟。如何为此设计喷嘴？我应该直接从spout中的MySQL查询，还是使用Kafka这样的消息队列？我想要实时性能，所以更喜欢第一种方式。但我发现很少有生产实现选择这种方式。为什么？
浏览 2提问于2014-07-08得票数 2
我有一个类似于的问题。不同之处在于，我使用使用kafka connect的SQL来读取topic中的记录。我以固定的间隔获取记录，但不知何故，我没有获得输出中的最后几条记录。例如，Kafka topic的最后一条记录的时间戳是2020-11-26T13:11:36.605Z，聚合值的最后一条时间戳是2020-11-26T12:59:59.999。我不明白为什么我没有得到topic中最后一条记录的聚合。请帮帮忙。这是我的代码。
sourceSQL = "CREATE TABLE flink_read_kafka (clientId INT, orderId INT, contactTim
浏览 0提问于2020-11-27得票数 0
我正在使用io.vertx.reactivex.kafka.client.producer.KafkaProducer客户端。客户端有一个rxWrite函数，它返回Single<RecordMetadata>。但是，我需要记录错误，如果有，在写操作。显然不是被处决。
我写了下面的工作例子。
test ()：测试链接和日志记录的函数
fun test(): Single<Int> {
val data = Single.just(ArrayList<String>().apply {
add("Hello")
浏览 0提问于2018-04-10得票数 0
回答已采纳
如果生成的记录失败，我希望将回调设置为被触发。最初，我只想记录失败的记录。
Confluent Kafka python库提供了添加回调的机制：
produce(topic[, value][, key][, partition][, on_delivery][, timestamp])
...
on_delivery(err,msg) (func) – Delivery report callback to call (from poll() or flush()) on successful or failed delivery
如何才能在kafka-python 中实现类似的行为
浏览 0提问于2017-09-24得票数 6
回答已采纳
我有一个生产者代码，在那里我发送消息给卡夫卡。没有任何失败或错误消息，代码可以执行，但不会发送消息。
kafka-python==2.0.1
Python 3.8.2
下面是我的代码，
from kafka import KafkaProducer
import logging
logging.basicConfig(level=logging.INFO)
producer = KafkaProducer(bootstrap_servers='127.0.0.1:9092')
producer.send('Jim_Topic', b'Message
浏览 3提问于2020-06-16得票数 0
假设我在向kafka发送数据时使用了Avro序列化。 在使用Apache Kafka的记录时，我同时获得了模式和记录。我可以使用模式来解析记录。我不明白为什么消费者会使用不同版本的模式来反序列化记录。有人能帮帮忙吗？
浏览 12提问于2020-09-24得票数 0
我有一个源连接器，它推动一个新创建的主题。"key.converter“和"value.converter”没有配置，因此转换使用Avro完成。
"key.converter.schemas.enable":"false"，"value.converter.schemas.enable":"false"，
当Kafka填充此主题时，它将在注册表中创建一个模式，其中主题名为“populates key”和“populates value”。例如，此架构可能具有如下配置的字段
"fields: [{"
浏览 1提问于2022-07-23得票数 0
1回答
我使用spring-cloud-sleuth-stream和spring-cloud-starter-stream-kafka发送span给kafka，连接时出现异常。spring cloud版本Edgware.SR1
spring
cloud:
stream:
kafka:
binder:
brokers: 192.168.1.177:9092
zkNodes: 192.168.1.177:2181
192.168.1.177是我的本地主机
我不明白为什么Updated
浏览 10提问于2018-02-05得票数 0
回答已采纳
Spring引导生成器应用程序向主题为== "AAA“的主题== "Topic0”发送消息，其中键为== "0“。使用者应用程序(也是spring-boot)有3个监听器/使用者，配置如下：
Topic0，分区== 0，Group0
Topic0，分区== 0，Group0
Topic0，分区== 0，Group1
public static final String KAFKA_TOPIC_0 = "Topic0";
public static final String KAFKA_GROUP_ID_0 = "Group0&
浏览 9提问于2022-04-10得票数 0
回答已采纳
我正在使用spring-kafka和spring-kafka-test版本1.0.2.RELEASE。
在我的一个测试中，我的应用程序使用KafkaTemplate和大多数默认配置设置将100条记录连续发送到EmbeddedKafka实例上的单个TopicPartion。
我使用KafkaTestUtils.getRecords(消费者)方法尝试从Kafka实例获取记录，并验证它们是否都已发送。
第一次调用getRecords时，我只收到一条记录。如果我再打一次我就能拿到剩下的99。
如果我显式地将消费者的位置设置为TopicPartition的开头，然后调用getRecords，我将得到全部
浏览 3提问于2016-08-12得票数 2
当我从spring.io创建一个新的项目时，它工作得很好。我可以在日志中看到span id、traceId和项目名称。但是当我尝试用现有的项目实现同样的功能时，用同样版本的spring boot和sleuth就不能工作了。应用程序中没有日志记录模式，这是spring提供的默认模式。在应用程序包级别和logging.level.org.springframework.web=INFO上，日志记录级别都设置为INFO。
浏览 154提问于2021-11-16得票数 0
1回答
当我检查使用者延迟时，它会显示某个特定的consumer-id是从某个主题的特定主机消费中运行的。但是当我去那个主机时，没有这样的消费者在运行。
如何杀死此consumer-id，以便可以为其所属的组重置使用者偏移量。
Kafka server version: 0.11.0.1
Kafka client version(stuck): 0.10.0.2
这个consumer-id被卡在了第一位，因为它无法使用消息，因为有些消息在Kafka中有标题。
我已经尝试过以下几种方法：
从不同的主机和不同的卡夫卡版本消费，它消费的信息，但消费者id，主机不会改变。
重新启动kafka代理，它
浏览 0提问于2019-02-18得票数 3
我收到了一些以卡夫卡结尾的事件。从这些事件中，我使用Kafka Streams应用程序获取id，并将其作为另一个主题中的一对(id，1)发送回Kafka。然后，我想看看id是否已经存在于ElasticSearch中，如果存在，则更新其计数器，否则在ElasticSearch中创建一个新记录，其中id来自Kafka，计数器设置为1，即将记录(id，1)向上插入到ES。
我希望使用Kafka Connect到ElasticSearch来实现这一点，但如果可能的话，这似乎并不是那么简单。我可以看到，将记录添加到ES是可行的，但与现有记录合并似乎是我还没有发现的事情。这已经可能了吗?如果可能，如何实现
浏览 4提问于2016-12-15得票数 3
我正在努力建立一个卡夫卡连接分布式模式应用程序，这将是一个卡夫卡到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
我的产品中有一个4节点的kafka集群，在这里我们使用自定义分区器，它执行id的mod 64来确定分区。自上周以来，在我们的一个节点上出现了不平衡的kafka messages_in速率，如所附图表所示。
。粉红行在kafka01节点上以速率表示消息，而蓝色黄线则在所有其他3个框中以速率显示消息。我使用datadog来监视和使用度量kafka.messages_in.rate。假设id分布没有变化，则在速率上消息的分布应该没有变化。我为调试该问题所采取的步骤包括
集群由4个节点中的每个节点上的16位领导者组成。
ISR在这4个盒子中也是平衡的，每个盒子都有32个ISR复制因子为2。
浏览 0提问于2018-04-02得票数 0
我在Kafka流中是新手，我的用例是将来自输入主题的每个单独传入记录的值与来自其单独先前记录的值进行比较，如果比较条件为真，则将包含比较结果的新记录与每个比较记录的索引发送到结果主题，否则不发送任何内容。(请注意，对于每个记录，所有传入记录都可能有一个唯一的键，也可能有一个空键)。
在Kafka消费者和生产者API中做这件事非常容易，但是((不使用外部DB来存储前面的记录)我只尝试使用(Kafka streams DSL API)，(其中包括KTable和KStream，以及它们的内部方法，如aggregate，reduce。等)，但可能是因为我是一个初学者，所以我没有在这些API中找到一种明
浏览 2提问于2018-12-23得票数 1
我正在使用MongoDB Kafka连接器，我可以成功地创建和更新记录，但无法删除连接器配置中的记录。我正在使用此配置
{
"name": "mongo-sink",
"config": {
"connector.class": "com.mongodb.kafka.connect.MongoSinkConnector",
"tasks.max": "1",
"topics": "test",
"connecti
浏览 1提问于2021-01-05得票数 0
1回答
我有一个Flink流应用程序，它正在消耗来自Kafka主题的数据，这个主题有3个分区。尽管应用程序一直在运行和工作，并且没有任何明显的错误，但我看到flink应用程序在所有3个分区上的消费者组都出现了滞后。
./kafka-consumer-groups.sh --bootstrap-server $URL --all-groups --describe
GROUP TOPIC PARTITION CURRENT-OFFSET LOG-END-OFFSET LAG CONSUMER-ID HOST CLIENT-
浏览 7提问于2022-06-17得票数 1
1回答
我们有一个非常简单的卡夫卡消费者(v2.6.2)。它是消费者组中的唯一使用者，而这个组是唯一一个从主题中读取信息的组(有6个分区，其中包含大约300万个事件)。代理也是2.6.x版本。
由于我们需要实现一个"excatly-once"场景，我们深入研究了一下，如果我们真的只使用了一次写入主题的事件。不幸的是，我们发现：--使用者有时跳过一个偏移量，有时甚至跳过一组分区的偏移量。
使用者所做的仅仅是日志记录。错误处理程序配置有日志记录。
@KafkaListener(id = "${spring.kafka.consumer.group-id}", topics
浏览 9提问于2022-02-28得票数 1
1回答
我有一个消费者，它消耗了一个话题。主题有6个分区。分配给组的单个使用者。我做投票，如下面的Consumer.poll(10000)，我退出消费者获取时，没有记录返回。
从文档中，我相信当没有要使用的记录和持续时间10000足以重新平衡和获取记录时，轮询返回为空。多数时间轮询使用所有分区的记录，但有时轮询从3个分区获取记录，然后返回空记录，而不消耗其他3个分区。
顺便说一下，我使用了2.0.1Kafka客户端，Kafka服务器版本是2.11-2.2.0。
任何人都知道为什么我的使用者跳过其他分区并返回空的records.what来使用所有分区。
浏览 1提问于2020-01-23得票数 1
回答已采纳
我正在开发一个包含两个微服务的应用程序: Microservice A将数据发送给kafka broker，Microservice B使用kafka的记录并将数据存储到mysql db中。我的目标是获得请求的全部跟踪，从接收POST请求的Microservice A开始，通过kafka并在Microservice B中结束，将在Microservice A中生成的Opentem遥测traceId传播到Microservice B日志中。Opentelemetry使用W3C标准来传播上下文。我能够在kafka记录头参数traceParent中插入corrent traceId
@Inject
浏览 7提问于2022-11-15得票数 0
1回答
目前，我们分析了一个可能的分布式跟踪工具，用于事件驱动的Microservice体系结构。
它现在看起来有点像下面的图片。
据我所知，与jaeger集成的所有应用程序(具有适当的jaeger依赖关系的spring引导应用程序)都将创建一个基于中间件的span (在我的例子中，主要是kafka )。
我的问题是，在调用链上，是否有一个调用的应用程序允许第三方API中没有jaeger跟踪集成。它会打破追踪链条吗？
或者，如果主应用程序链使用kafka在服务中进行通信，那么如何确保跟踪不被中断？
如果有人为此编写自定义代码，不幸的是，我在他们的文档中找不到答案。
浏览 4提问于2022-07-11得票数 0
我有一个编写批处理作业的用例 我需要阅读Kafka主题，并将数据记录到HDFS。我的代码如下所示 val df: DataFrame = spark.read
.format("kafka")
.option("subscribe", "test-topic")
.option("includeTimestamp", true)
.option("kafka.bootstrap.servers", "localhost:9092")
.option("group.id&
浏览 17提问于2019-02-05得票数 0
回答已采纳
来自MySQL和Kafka主题的事件数据。我们将从MySQL加载历史数据，然后切换到Kafka，以便当前的data.Mysql ID也会流入Kafka。因此，消息包含事件ID。Kafka流事件将使用spark作业进行处理，并最终将数据摄取到报表中。 问题:如何加载历史数据并实时切换到Kafka源。 1-创建临时使用者组以存储所有分区的当前偏移量和当前时间戳TS1。 2-从create date <= TS1所在的MySQL表中获取最小ID和最大ID。并通过500k记录的分区进行处理。 3-完成上述处理后，将时间戳传递到主spark作业，该作业将处理存储的偏移量中的streAming数据
浏览 18提问于2020-01-23得票数 0
我使用简单的kafka处理程序： @KafkaListener(
topics = Topic.NAME,
clientIdPrefix = KafkaHandler.LISTENER_ID)
public class KafkaHandler {
public static final String LISTENER_ID = "kafka_listener";
@KafkaHandler(isDefault = true)
@Description(value = "Event received")
public void onE
浏览 7提问于2021-10-15得票数 0
我在Server中有一个Person数据库，其中有大约20个表，如地址、许可证、亲属等。所有表都有每个人唯一的id参数。在这些表格中有数百万条记录。我需要结合使用普通id参数的人员的这些记录，并将其转换为具有某些列名更改的json表文件。这个json文件然后通过一个制片人被推送给卡夫卡。如果我能以kafka生产者作为项目编写者的例子--很好，但真正的问题是理解如何使用spring批处理项目阅读器、处理器和项目编写器来创建复合json文件的策略和细节。这是我的第一个Spring批处理应用程序，所以我对此还比较陌生。
我希望得到关于实现策略的建议，使用复合读取器或处理器使用person id作为游
浏览 1提问于2015-09-22得票数 0
有许多线程描述如何从指定的时间戳开始从Kafka获取记录。所以我认为Kafka‘知道’它存储的每一条记录的时间戳。我需要获取具有指定分区和偏移量的记录的时间戳。有可能吗？ 信息系统将错误的数据放到Kafka中(客户订单中的产品id错误)，我需要分析日志文件来找出原因。知道该记录的时间戳会容易得多。
浏览 32提问于2021-05-03得票数 0
我们使用Spring Kafka 2.2.2版本通过@KafkaListener和ConcurrentKafkaListenerContainerFactory从Kafka中检索记录。我们已经将max-poll-records配置为5，但是它始终只向消费者提供列表中的1条记录，而不是5条记录。
虽然使用相同的配置，但它可以在Spring Kafka 2.1.4版本中工作。
下面是我们的application.yml配置：
spring:
kafka:
bootstrap-servers: localhost:9092
consumer:
enable-auto-
浏览 0提问于2019-06-23得票数 1
3回答
您如何控制卡夫卡生产商或消费者的控制台日志记录级别？我正在Scala中使用Kafka0.9API。
每次调用KafkaProducer上的KafkaProducer时，控制台提供的输出如下所示。这是否表明我没有正确设置KafkaProducer，而不仅仅是过度日志记录的问题？
17:52:21.236 [pool-10-thread-7] INFO o.a.k.c.producer.ProducerConfig - ProducerConfig values:
compression.type = none
metric.reporters = []
metadata.max.age.ms
浏览 4提问于2016-02-10得票数 4
回答已采纳
正如我在标题中所说的，当生产者停止发送修补程序时，我想接收最后一条windowedBy消息。目前我正在手工操作，但首先，我要做一点描述。
我有一个卡夫卡制作者，它是从一个文件读取行(每一行是一个不同的jSon)，每一行是发送给卡夫卡与500毫秒的时间间隔。我只有120行(或jSons)。
我有一个消费者，它消耗了生产者发送的所有jSons。守则：
final KStream<String, Aggregate> transactions = builder.stream(kafkaProperties.getTopic(), Consumed.with(Serdes.Strin
浏览 1提问于2020-08-27得票数 2
我有一个星火结构流消费记录，从卡夫卡主题与2分区。
火花作业：2查询，每个查询来自两个单独的分区，运行于同一个火花会话。
val df1 = session.readStream.format("kafka")
.option("kafka.bootstrap.servers", kafkaBootstrapServer)
.option("assign", "{\"multi-stream1\" : [0]}")
.option(&#
浏览 0提问于2020-07-24得票数 1
我有Kafka主题中的数据(每10分钟发布一次数据)，我计划使用(批处理模式)读取这些数据，并将其推到MongoDB中。
请注意:这将使用GCP上的Composer/Airflow进行调度，这将创建一个Dataproc集群，运行星火代码，然后删除该集群。
以下是我的当前代码：
# read from Kafka, extract json - and write to mongoDB
df_reader = spark.readStream.format('kafka')\
.option("kafka.bootstrap.servers",kafk
浏览 6提问于2022-06-22得票数 -1
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
腾讯会议
实时音视频活动推荐
腾讯云开发者
