在Apache Kafka中暂停主题之前，无法获取发布的旧消息
在Apache Kafka中，暂停主题之前无法获取发布的旧消息。Kafka是一个分布式流处理平台，它以高吞吐量、可扩展性和持久性为特点，用于处理实时数据流。在Kafka中，消息被发布到主题(topic)中，消费者可以订阅这些主题并消费消息。
当一个主题被暂停(pause)时，意味着消费者将不再接收该主题的新消息。然而，暂停主题并不会影响已经发布到该主题的旧消息。已经发布的消息将继续保留在Kafka的日志中，直到它们被消费者消费或者根据配置的保留策略被删除。
暂停主题的场景通常是在需要对消费者进行维护或者升级时使用。通过暂停主题，可以确保消费者不会错过任何新消息，同时仍然可以访问已经发布的旧消息。一旦维护或升级完成，可以恢复主题，消费者将继续接收该主题的新消息。
对于Apache Kafka，腾讯云提供了一系列相关产品和服务。其中，腾讯云消息队列 CMQ 可以作为 Kafka 的替代方案，提供高可用、高可靠的消息队列服务。您可以通过腾讯云消息队列 CMQ 来实现类似 Kafka 的消息发布和订阅功能。更多关于腾讯云消息队列 CMQ 的信息，请访问以下链接：
腾讯云消息队列 CMQ 产品介绍：https://cloud.tencent.com/product/cmq
相关·内容
我有一个Kafka broker运行在一个多VM环境中(私有云-与我们自己的Kubernetes集群-4节点集群)。
我已经创建了一个spring引导应用程序，它的发行者必须发布一条消息到Kafka broker内部的Kafka主题。我让两个容器(Kafka broker & spring引导应用程序)运行在同一个Kubernetes集群中。
我无法访问卡夫卡代理(运行在同一个k8s集群中)，无法提供Kafka的服务名称:发布服务器的bootstrap.servers中的端口id
Spring的发行者配置：
props.put(ProducerConfig.BOOTSTRAP_SER
浏览 0提问于2020-08-18得票数 2
2017-08-31 10:15:20,715警告配置'internal.key.converter‘已经提供，但不是已知的配置。(org.apache.kafka.clients.admin.AdminClientConfig:231) 2017-08-31 10:15:20,715警告配置'status.storage.replication.factor‘已经提供，但不是已知的配置。(org.apache.kafka.clients.admin.AdminClientConfig:231) 2017-08-31 10:15:20,715警告配置'internal
浏览 3提问于2017-08-31得票数 3
回答已采纳
1回答
我最近已经将Confluent升级到4.1，但是模式注册似乎有一些问题。在confluent start模式注册中心(因此也是ksql-server)上，无法启动。
下面是模式注册日志中的错误：
[2018-04-20 11:27:38,426] ERROR Error starting the schema registry (io.confluent.kafka.schemaregistry.rest.SchemaRegistryRestApplication:65)
io.confluent.kafka.schemaregistry.exceptions.SchemaRegistryIn
浏览 2提问于2018-04-20得票数 0
回答已采纳
我试图运行一个管道，使用apache-束，源作为一个卡夫卡主题，目的地作为另一个卡夫卡主题。我已经编写了我的代码，并且运行良好(也就是说，在我认为的代码中没有错误)。但在输出主题中看不到数据--这是代码：
import apache_beam as beam
import apache_beam.transforms.window as window
from apache_beam.options.pipeline_options import PipelineOptions
from apache_beam.io.external.kafka import ReadFromKafka, W
浏览 8提问于2022-06-02得票数 -1
我对Springboot和Kafka非常陌生。在学校作业中，我们需要使用Springboot应用程序发布关于Kafka主题的Json数据。我的发布.java文件如下所示： import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.GetMapping;
import org.springframework.web.bind.an
浏览 27提问于2020-03-17得票数 0
我已经在我的Windows10笔记本电脑上在WindowsLinux3.4.12下安装了Kafka1.1.0和ZooKeep3.4.12。我可以在ubuntu中获取和使用消息，但是当我想从windows (使用java程序或使用kafka-console-Producer.bat工具)生成消息时，我有以下错误：
2018-05-11 15:31:01,449在用键向主题测试发送消息时出错: null，值: 15字节带有错误：(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.co
浏览 0提问于2018-05-11得票数 0
2回答
Apache
、、
我正在学习Apache Kafka (作为一种消息传递系统)，在这个过程中，我了解了术语StateStore，
我也知道Apache kafka streams，客户端API。
StateStore适用于消息传递系统上下文中的Apache kafka，还是适用于Apache Kafka Streams。
Apache是否有自己的StateStore实现或使用第三方实现(例如，rockdsb )。
有人能帮我理解这件事吗。
浏览 3提问于2019-12-28得票数 1
1回答
在使用ErrorHandlingDeserializer结合Avro处理错误时，我无法发布到Dlq主题。下面是发布时的错误。
主题TOPIC_DLT在60000 ms后不存在于元数据中。错误KafkaConsumerDestination{consumerDestinationName='TOPIC'，partitions=6，dlqName=‘TOPIC_DLT’}。容器-0-C-1 o.s.i.h.LoggingHandler:250 - org.springframework.messaging.MessageHandlingException:错误发生在消息处理程序or
浏览 3提问于2022-03-02得票数 0
回答已采纳
我正在尝试使用以下代码在动态创建的主题上发送消息：
@Service
public class KafkaMessageSender {
private static final Logger LOGGER = LoggerFactory.getLogger(KafkaMessageSender.class);
@Autowired
private KafkaTemplate<String, String> kafkaTemplate;
public void send(String topic, String payload) {
浏览 12提问于2018-02-13得票数 3
回答已采纳
我使用的是Storm版本1.1.0和Kafka版本0.10.1.2。
我创建Kafka-spout的方法如下：
public KafkaSpout<String, String> getKafkaSpout() {
String _kafkaBrokers = (String) props.get("bootstrap.servers");
String _topic = (String) props.get("kafka.topic.name");
String groupId = (String) props.get(&#
浏览 5提问于2017-10-19得票数 1
这是我第一次使用卡夫卡。我有一个spring引导应用程序，我正在使用kafka主题的消息，并将消息存储在DB中。我有一个处理DB fail的要求，如果DB关闭，则不应该提交该消息，并在一段时间后暂停消费消息，然后监听器可以再次开始使用消息。做这件事的更好的方法是什么。
i am using spring-kafka:2.2.8.RELEASE which is internally using kafka 2.0.1
浏览 9提问于2020-06-01得票数 1
Java: OpenJdk 11 Kafka: 2.2.0Kafka流库: 2.3.0
我正在尝试将我的Kafka应用程序部署到停靠器容器中，但是它在尝试使用TopicAuthorizationException创建内部状态存储时失败了。在当地效果很好。本地和服务器之间的主要区别是，它连接到部署了Kafka的服务器，并使用通常的Kerberos auth进行身份验证。我无法理解身份验证与本地存储之间的联系。
我的小溪看起来是这样的：
StreamsBuilder builder = new StreamsBuilder();
//We stream from the sour
浏览 2提问于2019-10-09得票数 8
回答已采纳
2回答
我刚接触Kafka，正在尝试实现一个简单的生产者，将数据发送到一个主题。如果该主题不存在，我希望将sutiation作为异常处理。
private Producer<UUID, Object> producer = createProducer();
private static Producer createProducer() {
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
"mybootstra
浏览 34提问于2019-08-08得票数 1
回答已采纳
我正在使用两个微服务，以便使用Kafka相互通信。在一个微服务上有侦听器，在另一个微服务上是发送者。问题是，每当生产者发送一条消息，消费者将得到20个重复的消息，我希望消费者只收到一条。
微型服务A：
@Service
public class LocationProducer {
private static final String MAINLOCATION = "mainlocation";
@Autowired
private KafkaTemplate<String,String> kafkaTemplate;
public void sendMes
浏览 10提问于2022-04-12得票数 3
我们增加了用于并行处理消息的分区数量，因为消息的吞吐量很高。一旦我们增加了分区的数量，订阅该主题的所有Streames线程就会死掉。我们更改了消费者组id，然后重新启动了应用程序，它工作正常。 我知道应用程序的changelog topic的分区数应该与源topic相同。我想知道这背后的原因。 我看到了这个链接- https://issues.apache.org/jira/browse/KAFKA-6063?jql=project%20%3D%20KAFKA%20AND%20component%20%3D%20streams%20AND%20text%20~%20%22partition%
浏览 23提问于2019-02-12得票数 8
回答已采纳
我是Apache Kafka的新手，我想知道apache Kafka中的消息可以有多大。如果消息的大小变得相当大，比如说几百MB，使用Apache Kafka是否有效？
我有一个场景，我想要将文件复制到HDFS以供Hadoop作业使用，这些文件也由其他进程使用。我在考虑首先将文件复制到Apache Kafka中，然后消费者可以将它们复制到HDFS，而其他消费者可以使用Kafka。这是不是最好的方法？
浏览 0提问于2015-04-29得票数 1
在附加新主题的消费者之前，我创建新主题并在apache kafka中生成第一条消息。然后附加新主题的消费者，但无法消费第一条消息。为什么..？
在本例中，已经记录了offset=1、提交的offset=1、lag=0。
“提交的offset=1”不就意味着它已经被消费了吗？我的问题是，为什么它已经被消费了。如果有什么不对劲的地方，请告诉我。
这是我的测试用例。
# create new topic
$ kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --replication-factor 1 --partition
浏览 0提问于2018-07-05得票数 0
我正在开发一个Spring引导应用程序。该应用程序充当Kafka-producer，其中生产/发布消息并不是很重要。我正在使用spring-kafka依赖项，其配置如下 @Bean
public ProducerFactory<String, Object> producerFactory() {
Map<String, Object> configProps = new HashMap<>();
configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, kafkaConfi
浏览 245提问于2021-08-12得票数 1
我们在项目中使用Kafka流状态存储，我们希望存储超过1MB的数据，但是我们得到了下面的例外：
序列化时消息为1760923字节，比使用max.request.size配置配置的最大请求大小还要大。
然后，我按照链接添加了以下配置：
topic.max.request.size=50000000
然后应用程序可以正常工作，但是当创建了状态存储内部主题时，但是当Kafka重新启动并且状态存储主题丢失/删除时，它可以正常工作，然后Kafka流处理器需要在启动应用程序时自动创建内部状态存储主题，此时它会抛出异常，即：
"Aorg.apache.kafka.streams.errors
浏览 3提问于2020-06-04得票数 1
我在这个上看到，我们可以使用org.apache.spark.streaming.kafka.KafkaUtils#createRDD在火花批处理作业中读取来自卡夫卡的消息，但是这种方法需要一个偏移范围，它需要一个“从偏移”和“直到偏移”。我从org.apache.spark.streaming.kafka.KafkaCluster#getLatestLeaderOffsets方法中获得了“from偏移量”，但是如何获得“直到偏移量”呢？我用的是kafka-2.1.1-0.9.0.1
浏览 0提问于2018-02-06得票数 0
1回答
我对mongo使用Kafka连接(源代码)，连接工作人员正在运行，但它不是将数据写入Kafka主题，而是使用源连接器和连接器的以下配置文件：
name=mongo-ff
tasks.max=1
connector.class =com.mongodb.kafka.connect.MongoSourceConnector
database=haifa
collection=alerts
key.converter = org.apache.kafka.connect.storage.StringConverter
value.converter = org.apache.kafka.connec
浏览 10提问于2020-01-22得票数 0
以下是我们正在使用的版本。
kafka - 2.10-0.8.2.2 flume - 1.7.0 zookeeper - 3.4.6
当我们尝试从Oracle数据库增量下载数据时，我一直收到以下错误
(kafka-producer-network-thread | producer-1) [ERROR -
org.apache.kafka.clients.producer.internals.Sender.run(Sender.java:130)]
Uncaught error in kafka producer I/O thread:
org.apache.kafka.common.pr
浏览 2提问于2017-04-25得票数 0
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
回答已采纳
在Kafka的消费者投票循环中，当poll方法抛出一个SerializationException时，有没有办法跳过这条消息(也就是“毒丸”)，继续消费主题中的下一个事件？
我可以捕获异常并使用consumer.seek()方法将偏移量移动到下一条消息，但该方法需要分区和偏移量作为输入参数。有没有办法获得这些值？
我在github代码库中有示例代码。要运行该示例：
$ git clone https://github.com/bjornhjelle/kafka-streams-examples-gradle.git
$ cd kafka-streams-examples-gradle
$ ./
浏览 0提问于2018-03-15得票数 4
3回答
我使用Spring 2.0.2.RELEASE和Spring (有效的pom显示了spring的2.1.6.RELEASE版本)。
我已经从使用常规的ByteArrayDeserializer到使用Confluent的反序列化器
properties.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, io.confluent.kafka.serializers.KafkaAvroDeserializer.class);
因此，我不需要获取字节，然后将它们反序列化为有效负载等等。但是副作用是一些旧消息-我不能再读了，因为它们的模式在汇合注册表
浏览 0提问于2018-08-17得票数 0
回答已采纳
我有一个流应用程序与下面的驱动程序代码的实时消息转换.
String topicName = ...
KStreamBuilder builder = new KStreamBuilder();
KStream<String, String> source = builder.stream(topicName);
source.transform(() -> new MyTransformer()).to(...);
KafkaStreams streams = new KafkaStreams(builder, appConfig);
streams.setUncaug
浏览 17提问于2017-02-22得票数 7
回答已采纳
我写了一个Java程序，它使用了Kafka库，我听说Kafka生产者有内部缓冲区来保存消息，这样它可以在以后重试。所以我创建了具有重试属性的幂等Kafka生成器。
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, System.getenv(KafkaConstants.KAFKA_URL));
props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer"
浏览 3提问于2018-01-22得票数 0
回答已采纳
当源主题分区计数= 1时工作正常。如果我将分区增加到任何大于1的值，我会看到下面的错误。既适用于低级，也适用于DSL API。有什么建议吗？可能会遗漏什么？
org.apache.kafka.streams.errors.StreamsException: stream-thread [StreamThread-1] Failed to rebalance
at org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:410)
at org.apach
浏览 3提问于2017-02-19得票数 6
回答已采纳
1回答
当我尝试使用Kerberized集群中的Spark流应用程序使用Kafka的消息时，我遇到了一些问题。我尝试了两种方法-- ：
基于接收机的方法：KafkaUtils.createStream
直接进场(无接收器)：KafkaUtils.createDirectStream
基于接收器的方法(KafkaUtils.createStream)引发两种类型的异常(不同的异常，无论我处于本地模式(--master local[*])还是纱线模式(--master yarn --deploy-mode client) )：
星火本地应用程序中奇怪的kafka.common.Brok
浏览 9提问于2017-12-26得票数 4
回答已采纳
6回答
了解消费群体id
、、
我重新安装了Apache 0.10.1.0。
我能够在命令提示符下发送/接收消息。
在使用Producer / Consumer示例时，我无法了解消费者示例上的group.id参数。
让我知道如何解决这个问题。
下面是我用过的消费者示例：
public static void main(String[] args) {
Properties props = new Properties();
props.put("bootstrap.servers", "localhost:9092");
浏览 4提问于2016-12-29得票数 42
我想知道使用java的消费群体的消费滞后。我试过使用 kafka-consumer-groups --describe --bootstrap-server localhost:9092 --group MyGroupName 并且滞后是可见的。 我如何在Java中做到这一点？ 我曾尝试使用org.apache.kafka.clients.admin.AdminClient，但无法获得每个消费者组的lag。 我正在使用confluent 5.0.1 which has kafka 2.0.1 org.apache.kafka - kafka-clients - 2.0.1
浏览 96提问于2019-05-09得票数 3
回答已采纳
我正在使用2.2.2.RELEASE(org.apache.kafka:kafka-clients:jar:2.0.1)和spring(2.1.1)。我无法执行事务，因为我的侦听器无法获得分配的分区。我为一次使用者创建了建议的配置。我正在尝试配置事务性侦听器容器，并且只需要一次处理
我使用事务管理器配置了生产者和使用者，用事务id配置了生产者，用isolation.level=read_committed配置了使用者。
@Bean(name = "producerFactory")
public ProducerFactory<String, MyObje
浏览 0提问于2019-07-22得票数 1
我试图使用KStreamBuilder将数据从一个主题移动到另一个主题。我尝试了以下代码，但有例外
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.common.serialization.Serdes;
import org.apache.kafka.streams.kstream.KStreamBuilder;
import org.apache.kafka.streams.KafkaStreams;
import org.apache.kafka.streams.Stream
浏览 1提问于2016-07-26得票数 1
2回答
我已经启动了我的zookeeper和Kafka服务器。我启动了我的Kafka制作器，它发送了10条主题为'xxx‘的消息。然后叫住了我的卡夫卡制作人。现在我启动了我的Kafka用户，并订阅了主题'xxx‘。我的消费者使用我的Kafka生成器发送的10条消息，该生成器现在没有运行。我需要我的Kafka消费者应该只消费运行Kafka服务器的消息。有什么方法可以做到这一点吗？以下是我的消费者属性中的内容。
props.put("bootstrap.servers", "localhost:9092");
String consumeGroup
浏览 1提问于2016-09-30得票数 3
我正在使用一个非事务性生产者，并试图了解在成功/失败场景中如何处理回调。
为了发送成功，我看到回调由kafka生产者-网络线程(“已发送确定”消息)执行。
成功发送消息-卡夫卡-生产者-网络-线程00:59:17.522
00:59:16.850 [org.springframework.kafka.KafkaListenerEndpointContainer#0-0-C-1] INFO o.a.kafka.common.utils.AppInfoParser - Kafka version: 2.3.1
00:59:16.858 [org.springframework.kafka.Kaf
浏览 7提问于2020-03-15得票数 0
回答已采纳
根据Apache的文档，Kafka在其本地API的通信中使用了二进制TCP协议，但它们也为不支持Apache的本地API的语言提供了基于URL的restful。我只是想知道本地二进制TCP协议(在本机API中支持)是否比基于restful的与代理节点的通信有什么好处？我还在想restful是否还会维护only once属性呢？
编辑: restful指南在这里：，它解释了如何在restful上生成和使用Kafka的消息
浏览 1提问于2017-09-24得票数 2
回答已采纳
我知道Apache MM1和Apache MM2之间有什么不同。
Kafka MM1不支持活动主动设置和偏移同步，这也是MM1和更多的问题之一。
但是我无法理解复制器和集群链接之间的区别。
浏览 13提问于2022-05-05得票数 0
回答已采纳
当没有领袖时，我无法提供信息。我使用以下命令创建了具有复制因子2的新主题
~/kafka/bin/kafka-topics.sh --create --zookeeper localhost:2181 --topic TestTopic202 --partitions 1 --replication-factor 2
然后
~/kafka/bin/kafka-topics.sh --describe --zookeeper localhost:2181 --topic TestTopic202
输出：
OpenJDK 64-Bit Server VM warning: If the num
浏览 0提问于2019-09-03得票数 0
回答已采纳
我正在为Apache使用Spring，并创建了一个服务，该服务通过Spring的org.apache.kafka.clients.producer(通过Spring的KafkaTemplate )使用卡夫卡生产者( Kafka Producer，KafkaTemplate)向主题发送消息。在目标Kafka集群上，我禁用了自动主题创建。使用这里列出的生产者配置的组合( )，我成功地控制了重试请求的次数、重试之间的时间等等。
如果我提供了一个不存在的主题，那么当我期望它达到max.block.ms).值时，请求就会超时。但是，超时之后，我将继续在retry.backoff.ms设置的时间间隔内获
浏览 4提问于2020-02-17得票数 2
回答已采纳
5回答
我试图阅读有关卡夫卡主题的信息，但我无法阅读。这个过程会在某个时候被终止，而不会读取任何消息。
以下是我得到的再平衡错误：
[2014-03-21 10:10:53,215] ERROR Error processing message, stopping consumer: (kafka.consumer.ConsoleConsumer$)
kafka.common.ConsumerRebalanceFailedException: topic-1395414642817-47bb4df2 can't rebalance after 4 retries
at kafka.c
浏览 3提问于2014-03-21得票数 18
回答已采纳
我无法让下面的代码工作：
"use strict";
let kafka = require('kafka-node');
var conf = require('./providers/Config');
let client = new kafka.Client(conf.kafka.connectionString, conf.kafka.clientName);
let consumer = new kafka.HighLevelConsumer(client, [ { topic: conf.kafka.readTopic } ],
浏览 1提问于2016-03-08得票数 3
回答已采纳
根据Apache Kafka文档，消息的顺序可以在分区内实现，也可以在主题中的一个分区内实现。在这种情况下，我们得到的并行性好处是什么，它等同于传统的MQs，不是吗？
浏览 11提问于2015-04-23得票数 62
我使用的是kafka高级消费者。当我启动消费者时，它会找到所有新的消息。当我使用Java kafka生成器生成新消息时，它会发现这些消息。然而，一分钟后，它继续循环，但没有找到新的消息。当我在调试器中暂停执行时，使用者突然开始查找要使用的消息。我在Java中使用的是0.8.0版本。请注意，在发生错误时，使用消息的进程将在单独的“错误”主题中生成消息。当我停止生成这些错误消息时，我就不再遇到这个问题。
浏览 0提问于2014-12-09得票数 1
1回答
我可以针对Kafka location安装在命令行上发送和接收消息。我还可以通过Java代码发送消息。这些消息显示在Kafka命令提示符中。我还有一个用于Kafka消费者的Java代码。代码昨天收到消息。然而，它今天早上没有收到任何消息。代码未更改。我想知道属性配置是不是不太正确。下面是我的配置：
制片人：
bootstrap.servers - localhost:9092
group.id - test
key.serializer - StringSerializer.class.getName()
value.serializer - StringSerializer.class.ge
浏览 4提问于2016-06-24得票数 0
我们使用Apache Kafka在我们的开发环境中执行负载测试。
在我们的Linux机器中，我们安装了confluent kafka，空间有限，因此为了执行负载测试，我们在主题中添加了retention.ms属性。
我的想法是在消息被消费者消费后从主题中删除。
我试过了
kafka-topics --zookeeper localhost:2181 --alter --topic myTopic --config retention.ms=10000
它不起作用，因此我们重新创建了主题，并尝试了下面的选项。
kafka-configs --alter --zookeeper localhost
浏览 1提问于2020-05-21得票数 0
我使用的是Kafka 0.9.0.1。
我第一次启动应用程序时，需要20-30秒才能检索到主题中的“最新”消息
我使用过不同的Kafka代理(具有不同的配置)，但我仍然看到这种行为。对于后续的消息，通常不会出现速度缓慢的情况。
这是预期的行为吗？通过运行此示例应用程序并将代理/主题名称更改为您自己的设置，您可以在下面清楚地看到这一点
public class KafkaProducerConsumerTest {
public static final String KAFKA_BROKERS = "...";
public static final Stri
浏览 0提问于2016-07-23得票数 2
我有3个spring消费者(同一组)从3个分区接收消息。我想检测这些使用者中的一个何时停止从一个分区读取(其他两个消费者继续从其他两个分区读取)。到目前为止，这种情况已经发生过两次，一旦检测到，就很容易通过重新启动所有消费者来修复，从而导致重新平衡。问题是，在这两种情况下，如果早点知道是很好的。所以我试着用ListenerContainerIdleEvent就像这样-
@EventListener
public void eventHandler(ListenerContainerIdleEvent event) {
LOG.info("idle event fired! li
浏览 0提问于2017-12-09得票数 0
回答已采纳
我正在使用KakfkaStreams (2.3.0)来使用流和Ktable (压缩主题)之间的连接来丰富价值。问题是压缩后的主题被写入到使用流主题的不同分区器中，因此联接无法按预期工作(一些键不匹配，因为它们位于不同的分区中)。
代理版本为0.10.2。
我开始研究如何重新划分压缩的主题，方法是使用一个groupBy()，后面跟着一个()，但是当它开始读取创建的重新分区主题时，它开始抛出一个带有消息的：
输入记录ConsumerRecord(主题=映射-表-重新分区，分区= 18，leaderEpoch =空，偏移量= 0，CreateTime = -1，序列化键大小= 37，序列化值大小=
浏览 1提问于2019-08-28得票数 0
回答已采纳
使用汇合5.4.1
我们碰巧在将连接流从KTable KTable转发到另一个主题时，在新的KTable外键连接中遇到了一个问题。
模式注册表和Avro序列化器相关构件:合流版本5.4.1 kafkastreams客户端: 5.4.1-ccs
外键跨ktable连接之后，虽然使用“for each”从ktable消费流，但如果使用“to”将联接流转发到另一个主题，则会引发模式注册表错误。
将错误中提到的模式与在模式注册表中注册的模式进行比较，它是完全相同的…。似乎卡夫卡2.4.0已经有了类似的问题：和这个问题继续存在，同时将它转发到另一个主题
Caused by: io.confluent.k
浏览 5提问于2020-03-16得票数 0
回答已采纳
当Kafka Consumer在属性中具有事务性语义时，它不能消费任何消息。但当我删除该属性或将该属性更新为read_uncommited时，它的消费消息。
以下是我的Kafka消费者属性：
Properties props = new Properties();
props.put("bootstrap.servers", "10.2.200.15:9092");
String consumeGroup = "cg3";
props.put("group.id", consumeGroup);
/
浏览 0提问于2018-01-25得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
