Apache Kafka -生产者-生产者之间的等价性消息
Apache Kafka是一种高性能、可扩展、分布式流式数据平台，用于处理和传输大规模的实时数据流。它采用发布-订阅模式，允许数据的高吞吐量和低延迟传输。
生产者是Kafka中的一个角色，用于将数据推送到Kafka集群。生产者可以将消息发送到Kafka的一个或多个主题（Topic），并且可以选择将消息分区（Partition）到不同的分区中。生产者之间的等价性消息指的是在分区复制的情况下，生产者能够确保在发送消息时保持消息的顺序性和一致性。
等价性消息保证了消息的有序性和可靠性，确保了生产者发送的消息不会丢失或被重写。通过Kafka的复制机制，每个分区的副本都会被分配给不同的Broker节点，以保证故障时的数据冗余和高可用性。
生产者之间的等价性消息在以下场景中非常有用：
- 保证消息的顺序性：生产者可以将相关的消息发送到同一个分区，以确保消费者按照正确的顺序消费消息。
- 提供冗余和容错性：通过将消息复制到多个分区，即使其中一个生产者或分区发生故障，仍然可以从其他分区获取数据，确保数据的可靠性和高可用性。
- 实现负载均衡：通过将消息均匀地分布到多个分区，可以实现生产者之间的负载均衡，避免某个生产者过载而影响整体性能。
对于Apache Kafka，腾讯云提供了一系列相关产品和解决方案：
- 腾讯云消息队列CMQ：https://cloud.tencent.com/product/cmq
- 腾讯云云原生消息队列TDMQ：https://cloud.tencent.com/product/tdmq
- 腾讯云消息队列Ckafka：https://cloud.tencent.com/product/ckafka
这些产品提供了高可用性、低延迟和高吞吐量的消息传递服务，可以满足不同规模和场景下的实时数据处理需求。
相关·内容
我有两个Kafka代理:server2 1:9092和server2 2:9092--我使用Java客户机向这个集群发送消息，这是代码：
@Test
public void sendRecordToTopic() throws InterruptedException, ExecutionException {
//See at http://kafka.apache.org/documentation.html#newproducerconfigs
Properties props = new Properties();
props.put(ProducerConf
浏览 0提问于2015-11-02得票数 0
回答已采纳
看起来，重复是由Kafka broker通过在分区级别跟踪消息来执行的：
消息序列号
生产者id
我所读到的每一篇文章都谈到了这一问题，解决了生产者或代理错误导致生产者重新尝试发送的问题。那当制片人倒下的时候呢？生产者id是由我控制的静态id，还是每次注册生产者节点时由代理重新分配？如果重新分配了生产者id，并且与重新启动之前不同，就会发生重复，对吗？
我不明白为什么他们会这样设计生产者id，但我在org.apache.kafka.clients.producer.ProducerConfig中找不到任何的org.apache.kafka.clients.producer.Pr
浏览 0提问于2019-05-09得票数 1
回答已采纳
我使用的是反应性- kafka (akka kafka流)：
什么是相当于下面的代码使用反应性-卡夫卡？
import org.apache.kafka.clients.consumer.KafkaConsumer
...
val properties = new Properties()
properties.put("bootstrap.servers", "kafka:9092")
val kafkaConsumer = new KafkaConsumer[String, String](properties)
kafkaConsumer.partit
浏览 6提问于2017-10-02得票数 0
回答已采纳
可以使用JMeter向Apache Kafka推送消息吗？
如何实现producer ( JAVA)向Kafka推送消息。
致敬，阿南德
浏览 6提问于2015-03-30得票数 1
我正在使用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
我有一个卡夫卡环境，其中有3 brokers和1 zookeeper。我已经在我的主题中推送了超过20K的消息。Apache Storm计算producer添加的topic中的数据。
几个小时过去了，当我试图向kafka发送消息时，它显示了以下异常
org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms.
重启kafka服务器后，它工作正常。但在production上，我不能每次都重新启动服务器。所以有人能帮我解决我的问题吗？
我的kafka配置如下：
prodPrope
浏览 0提问于2016-06-03得票数 6
我是Apache Kafka的新手，我正在尝试配置Apache Kafka，使其尽可能多地接收来自生产者的消息，但它只在每个特定时间向消费者发送配置数量的消息。
浏览 0提问于2020-12-23得票数 1
我使用的是Publish Kafka处理器的下一个配置：
编辑kafka和zookeeper配置：
zookeeper.properties
authProvider.sasl=org.apache.zookeeper.server.auth.SASLAuthenticationProvider
requireClientAuthScheme=sasl
jaasLoginRenew=3600000
zookeeper_jaas.conf
Server {
org.apache.zookeeper.server.auth.DigestLoginModule required
us
浏览 3提问于2021-01-14得票数 1
生产者api( Kafka -client-1.0.1)无法在我托管在google云中的远程Kafka代理(v1.1.0)中生成消息。
我已经配置了server.properties文件，添加了以下内容：
listeners=PLAINTEXT://0.0.0.0:9092
advertised.listeners=PLAINTEXT://<Local-IP>:9092
但结果是一样的。
我的制作人Api看起来像这样：
public class KafkaProducerExample {
private final static String TOPIC = "my-t
浏览 0提问于2018-04-07得票数 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
我一直在使用Java创建一个Kafka生产者示例。我一直在发送正常的数据，这只是“测试”+整数作为价值卡夫卡。我使用了下面的属性，在启动了生产者客户端之后，消息正在传递，在此过程中，我会杀死代理，然后突然收到下面的错误消息，而不是重新尝试。
使用具有3个分区和复制因子的3个代理和主题作为3，而不使用min-insync-副本。
下面的是配置了 config.put(ProducerConfig.ACKS_CONFIG，“all”)的属性；
config.put(ProducerConfig.MAX_IN_FLIGHT_REQUESTS_PER_CONNECTION，“1”；config.pu
浏览 3提问于2019-01-09得票数 1
我试着用kafka生产者发送java字符串消息。字符串消息是从Java中提取的。
JavaPairDStream<String, String> processedJavaPairStream = inputStream.mapToPair
(record-> new Tuple2<>(record.key(), record.value())).mapValues(message -> message.replace('>', '#'));
String outTopics = &#
浏览 9提问于2019-09-18得票数 0
回答已采纳
2回答
我在视频教程中看到，Kafka Broker在producer发布消息时支持3种类型的确认。
0-开火后忘记
1- Leader Ack
2-所有代理的Ack
我正在使用Kafka的Java API来发布消息。这是必须使用特定于每个代理的server.properties为每个代理设置的内容，还是必须由生产者设置的内容？如果必须由生产者设置，请解释如何使用Java API进行设置。
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerRe
浏览 0提问于2017-07-09得票数 13
回答已采纳
2回答
我对Kafka2.6.0中的消息大小配置有点困惑。但让我们来说说故事：
我们正在使用一个由3个节点组成的Kafka集群。到目前为止，对于消息的标准配置。"zstd压缩“被激活。
相关的代理配置很简单：
compression.type=zstd
在这一点上，生产者配置也很简单：
compression.type=zstd
现在，我们希望将具有8M字节的消息放到特定的主题中。此数据的压缩大小仅为200千字节。
如果我将这些数据放入主题中，就会发生以下错误：
sudo /opt/kafka/bin/kafka-console-producer.sh --topic XXX --produc
浏览 5提问于2020-11-05得票数 5
回答已采纳
当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
1回答
Spring循环发送消息
、、、
我使用spring依赖版本2.8.1，配置如下：
kafka:
bootstrap-servers: localhost:9092
producer:
acks: all
key-serializer: org.apache.kafka.common.serialization.StringSerializer
value-serializer: org.apache.kafka.common.serialization.StringSerializer
buffer-memory: 16384
我试图向一个主题发送大约65.
浏览 7提问于2022-02-08得票数 1
KafkaStream在很长一段时间后无法生成数据。(超过设定的过期时间)
在记录错误消息后，甚至KafkaStream也死了。
例外情况如下：
org.apache.kafka.common.KafkaException: Cannot execute transactional method because we are in an error state
at org.apache.kafka.clients.producer.internals.TransactionManager.maybeFailWithError(TransactionManager.java:784)
at o
浏览 0提问于2018-12-30得票数 2
如何使火花写入消息到卡夫卡中的所有分区，以便我可以使用一个直接流，并提高流的性能。
这是我的密码：-
object kafka {
def main(args: Array[String]) {
val sparkConf = new SparkConf().setAppName("FlightawareSparkApp")
sparkConf.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
val ssc = new Streamin
浏览 2提问于2017-04-26得票数 1
我有两台机器localhost和192.168.1.110来运行两台独立的单机kafka。
(1)在localhost上，我运行：
主控台kafka_2.11-0.10.0.0bin/kafka-➜-Producer.sh --broker-list 192.168.1.110:9092 --topic test这是一条消息2016-08-24 18:15:27,441向topic test发送消息时出错: null，value: 2字节错误:包含1条记录的(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) or
浏览 13提问于2016-08-24得票数 0
3回答
我是卡夫卡的新手。我有关于kafka配置的问题。
我想使用如下所示的独立服务器，
server1:卡夫卡生产者server2:卡夫卡经纪人，卡夫卡消费者，动物园管理员
但是，我不能向broker发送消息。我收到了这个错误消息。
在控制台生产者(Server1)上，控制台标准输出错误消息`
[2016-05-24 16:41:11,823] ERROR Error when sending message to topic twitter with key: null, value: 3 bytes with error: Failed to update metadata after 6000
浏览 1提问于2016-05-25得票数 6
我使用EmbeddedKafkaBroker编写了一个简单的测试，我创建了一个测试生成器并发送了一条消息，但是我的KafkaListener没有被触发，所以每次测试都失败。有没有办法测试我的卡夫卡消费者，这样我就可以确保测试代码的覆盖范围？我希望我的假生产者(producerTest)从测试类中触发我的“真正的”卡夫卡消费者，并处理这些消息。
卡夫卡消费者：
@Component
@EnableAutoConfiguration
@Slf4j
public class MyKafkaListener {
@KafkaListener(topics = "${kafka.top
浏览 20提问于2022-07-27得票数 0
为什么kafka 0.10控制台生产者不能向kafka 0.9发送消息？
我在0.9版本中启动了kafka控制台消费者(在服务器端)。我正在启动0.10版本的kafka控制台生产者(在客户端)。然后我得到了producer exception (下图)。
为什么新的生产者不能向旧的消费者发送消息？是否可以将新的生产者设置为一种弃用模式，以便向旧的kafka发送消息？
$ my_topic \windows\kafka-console-producer.bat --broker-list my_server:9092 --主题bin
2016-08-30 14:03:53,365错误kafka生
浏览 11提问于2016-08-30得票数 0
回答已采纳
我是卡夫卡的新手。我尝试通过java应用程序发送消息并在命令行提示符中使用它，但该消息未在CLI上显示。
下面是java代码：
package com.kafka.prj;
import java.io.BufferedReader;
import java.io.InputStreamReader;
import java.util.Properties;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.KafkaProducer;
浏览 0提问于2016-06-09得票数 1
我对FlinkKafkaConnector有困难。
虽然我已经导入了org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011，但是我得到了以下错误。
Too many arguments for constructor FlinkKafkaProducer011
根据的说法，这些论点是正确的。但是，我的IntelliJ显示了这个定义。
public FlinkKafkaProducer011() {
}
有人知道怎么修吗？
更新
这是IDE的定义。
public class FlinkKafkaProducer
浏览 5提问于2020-08-11得票数 0
回答已采纳
2回答
val props = new Properties()
props.put("bootstrap.servers", "foo:9092,bar:9092")
props.put("acks", "all")
props.put("retries", 1)
props.put("batch.size", 10000)
props.put("linger.ms", 5)
props.put("key.serializer", "org.apache.kafka
浏览 0提问于2016-11-13得票数 2
1回答
在中尝试使用时间戳时，我发现了一些奇怪的东西。在从生产者发送了几条消息之后，我运行了kafka-console-consumer.sh并验证了这些消息是否在主题中。我叫住制片人，等了一分钟。当我重新运行kafka-console-consumer.sh时，它没有显示我之前生成的消息。我还添加了producer.flush()和producer.close()，但结果仍然相同。
现在，当我停止使用时间戳字段时，一切都正常工作，这让我相信使用时间戳的消息有一些挑剔之处。
我使用的是Kafka_2.11-2.0.0 (2018年7月30日发布)
以下是示例代码。
import org.apache.
浏览 19提问于2019-11-23得票数 1
在.yaml文件中，我们设置了
spring.cloud.stream.kafka.binder.configuration.enable.idempotence as true.
现在，当应用程序启动时，我们可以看到如下日志
[kafka-producer-network-thread | test_clientId] org.apache.kafka.clients.producer.internals.TransactionManager - [Producer clientId=test_clientId] ProducerId set to 0 with epoch 0
当向主题生成
浏览 147提问于2021-03-12得票数 1
回答已采纳
我的Apache Kafka生产者(0.9.0.1)间歇性地抛出一个
org.apache.kafka.common.errors.NotLeaderForPartitionException
我执行Kafka发送的代码如下所示
final Future<RecordMetadata> futureRecordMetadata = KAFKA_PRODUCER.send(new ProducerRecord<String, String>(kafkaTopic, UUID.randomUUID().toString(), jsonMessage));
try {
浏览 1提问于2016-04-28得票数 10
回答已采纳
我注意到，自从Kafka 0.8.2.0以来，Kafka附带了一个新的maven模块：
<!-- https://mvnrepository.com/artifact/org.apache.kafka/kafka-clients -->
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
<version>0.8.2.0</version>
</
浏览 0提问于2018-07-21得票数 2
回答已采纳
在从客户端(它连接到包含基于SSL的身份验证的服务器端)进行生产时，我收到以下警告：
[2016-03-10 07:09:13,018] WARN The configuration ssl.keystore.location = /etc/pki/tls/certs/keystore-hpfs.jks was supplied but isn't a known config. (org.apache.kafka.clients.producer.ProducerConfig)
[2016-03-10 07:09:13,019] WARN The configuration ssl.
浏览 1提问于2016-03-10得票数 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
回答已采纳
1回答
错误
Expiring 2 record(s) for catering-0:120012 ms has passed since batch creation
首先我使用了配置
zookeeper:
image: wurstmeister/zookeeper
container_name: zookeeper
ports:
- "2181:2181"
restart: always
kafka:
image: wurstmeister/kafka
container_name: kafka
浏览 9提问于2022-07-11得票数 0
1回答
如何处理卡夫卡中的大消息，比如超过20 to的消息等等。
2009-03- 13 08:59:10,923错误当发送消息到主题测试时，键:13字节，值: 11947696字节与错误：(org.apache.kafka.clients.producer.internals.ErrorLoggingCallback) org.apache.kafka.common.errors.RecordTooLargeException:请求包含一条大于服务器将接受的最大消息大小的消息。
2019-03-1303:59:14,478在用键向主题测试发送消息时出错: 13字节，值: 11947696字节带有错误
浏览 0提问于2019-03-13得票数 1
回答已采纳
我写了一个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
回答已采纳
Kafka客户端: 0.11.0.0-cp1 Kafka Broker：
在Kafka broker滚动重启时，我们的应用程序在发送到broker时丢失了一些消息。我相信滚动重启应该不会丢失任何消息。下面是我们正在使用的producer (使用Producer和异步send()，而不使用callback/future等)设置：
val acksConfig: String = "all",
val retriesConfig: Int = Int.MAX_VALUE,
val retriesBackOffConfig: Int = 1000,
val batchSize: In
浏览 0提问于2019-03-20得票数 0
我不能让这个起作用。我正在尝试将一条简单的测试消息推送到kafka主题。以下是代码和错误
Properties props = new Properties();
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "kafkaBrokerfqdn:6667");
props.put(CommonClientConfigs.SECURITY_PROTOCOL_CONFIG, "SASL_PLAINTEXT");
props.put(ProducerConfig.ACKS_CONFI
浏览 3提问于2018-04-17得票数 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
1回答
星火流中的卡夫卡再利用
、、、、
我们有一个火花流应用程序(下面是代码)，它从kafka获取数据，并在将数据插入到MongoDB之前(对每条消息)进行一些转换。我们有一个中间件应用程序，它将消息(批量)推入Kafka，并等待火花流应用程序的确认(针对每条消息)。如果中间件在将消息发送到Kafka后的一段时间(5秒)内没有接收到确认，中间件应用程序将重新发送该消息。火花流应用程序能够接收大约50-100条消息(一批)，并在5秒内发送所有消息的确认。但是，如果中间件应用程序推送超过100条消息，则由于火花流发送确认的延迟，导致中间件应用程序重新发送消息。在当前的实现中，每次要发送确认时，我们都要创建生产者，这需要3-4秒。
pac
浏览 0提问于2017-08-13得票数 3
回答已采纳
我想聚合3个主题，但只是为了聚合策略的PoC，我尝试聚合2个主题。 因此，我已经配置了以下Camel路由： @Override
public void configure() throws Exception {
CamelContext contexto = new DefaultCamelContext();
try{
onException(SQLException.class, Exception.class)
.handled(true)
.maximumRedeliveries(1)
.process
浏览 44提问于2020-08-15得票数 1
回答已采纳
需要帮助理解Kafka中的'acks‘属性，该属性在生产者配置中。 假设我错过了在producer类中传递'acks‘属性。然后，默认情况下，Kafka认为它是‘ack =0’。现在假设我需要有'acks=all‘，这样它才能保证我的消息传递到Kafka broker。有没有什么方法可以确保server.properties文件中的任何设置都是默认的‘acks=all’？ 我已经阅读了Kafka文档https://kafka.apache.org/documentation/
浏览 8提问于2019-09-19得票数 1
Apache如何管理集群中不同代理之间的同步时间戳，哪个时间实际上被记录并转换为消息创建时间戳，Kafka如何验证它的时间精度
浏览 3提问于2021-11-17得票数 0
回答已采纳
1回答
我正在尝试将定制的db数据(从Microsoft表中派生，在其他表上连接->是如何到达的)发布到Kafka集群。
我能够发布和使用来自Apache NiFi - to /- Apache Kafka的消息。
但我无法在合流云中发布来自Apache NiFi到Kafka的消息。
是否可以使用创建的API键发布/消费来自Apache NiFi (Server)的消息到合流云？
如果是，Apache的PublishKafkaRecord处理器和ConsumeKafkaRecord处理器中对应的属性是什么？
如果没有，请分享任何其他的想法来克服这个限制。
浏览 8提问于2022-07-27得票数 0
我正在测试使用Apache在Kafka (0.8.2.1)上发送消息的简单生产者。我在camel中使用java创建了端点。
CamelContext ctx =new DefaultCamelContext();
PropertiesComponent properties=new PropertiesComponent();
properties.setLocation("com/camel/test/props.properties");
ctx.addComponent("properties",properties);
final String ur
浏览 1提问于2015-07-07得票数 4
回答已采纳
在使用Apache Kafka这样的事件/消息代理来解耦两个系统时，我遇到了一个问题。该问题与前端在后端触发动作有关：
如果发布者既不知道“身份”，也不知道消费后端的数量，那么生产者(前端服务)如何知道发布的事件已经被所有后端服务(作为消费者)正确处理？
准确地说:用户可以使用前端UI更改他们的电子邮件地址。关联的服务将该“更改请求”事件发布到Kafka中的适当主题。然后，UI窗体被“锁定”，以防止后续的更改请求，直到每个使用者都完全处理了更改事件。但目前还不清楚如何检测这种状态。
浏览 0提问于2018-08-15得票数 2
1回答
我正试着思考卡夫卡连接( Kafka )是如何工作的，但我无法理解一件特别的事情。
从我的阅读和观察来看，我理解Kafka Connect允许您使用Source连接器将数据发送到Kafka，使用Sink连接器从Kafka读取数据。最重要的是，Kafka从某种程度上抽象出了所有特定于平台的东西，而你所关心的就是有合适的连接器。例如，您可以使用PostgreSQL源连接器向Kafka写入，然后使用Elasticsearch和Neo4J Sink连接器并行读取卡夫卡的数据。
我的问题是:这种抽象是如何工作的？为什么不同人编写的源连接器和Sink连接器能够一起工作？为了从Kafka读取数据并将其写入任
浏览 2提问于2021-02-05得票数 0
下面是我的生产者配置，如果您看到他们的压缩类型是gzip，即使我提到了为什么消息没有发布并且失败的原因是压缩类型
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, edi856KafkaConfig.getBootstrapServersConfig());
props.put(ProducerConfig.RETRIES_CONFIG, edi856KafkaConfig.getRetriesConfig());
props.put(ProducerConfig.BATCH_SIZE_CONFIG, edi8
浏览 0提问于2017-12-07得票数 1
我们现在正在将我们的一项服务从通过传统通信技术推送数据转移到Apache Kafka。
当前的逻辑是向IBM MQ发送一条消息，并在出现错误时重试。我想重复这一点，但我不知道在这种情况下代理提供了什么保证。
假设我通过producer通过Java客户端库批量发送了100条消息。假设它到达集群，是否有可能只接受其中的一部分(例如，磁盘已满，或者我在写操作中接触到的一些分区复制不足)？我可以从我的生产者那里检测到这个问题，然后只重试那些不被接受的消息吗？
我搜索了kafka atomicity guarantee，但一无所获，可能有一个广为人知的术语
浏览 0提问于2019-07-31得票数 1
我试图使用+ Kafka绑定对Apache的“精确一个交付”概念做一些PoC。
我安装了Apache " Kafka _2.11-1.0.0“并在生产者中定义了"transactionIdPrefix”，我理解这是我在Spring中启用事务所需要做的唯一事情，但是当我这样做并在同一个应用程序中运行简单的Source & Sink绑定时，我看到一些消息在使用者中被接收和打印，而有些则会出错。
例如，收到消息#6：
[49] Received message [Payload String content=FromSource1 6][Headers={kafka_offs
浏览 0提问于2017-12-27得票数 0
我正在尝试使我的应用程序使用模式注册中心从confluent开始工作，但此时我并不完全控制生产者，甚至可以将它们看作是遗留的应用程序，这些应用程序根本不绑定到汇流产品。
我在查看汇合信息，似乎所有的消息都应该包含在有效负载中，一个Magic和Schema
否则，当我试图使用它时，我会得到一个错误：
[2020-09-25 13:12:09,008] ERROR WorkerSinkTask{id=s3_parquet_connector-0} Task threw an uncaught and unrecoverable exception (org.apache.kafka.connec
浏览 5提问于2020-09-28得票数 0
回答已采纳
1回答
我正在尝试使用Apache Flink 1.6.0从kafka主题中读取一些消息，对它们进行转换，最后将它们发送到另一个kafka主题。我使用保存点保存应用程序的状态，以备取消和重复使用。问题是我在重启后读取消息时出现了重复。kafka版本为011。感谢你的任何有用的评论。
浏览 5提问于2018-09-27得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
即时通信 IM活动推荐
腾讯云开发者
