如何在reactor-kafka中重试失败的ConsumerRecord
在reactor-kafka中，可以通过配置重试策略来处理失败的ConsumerRecord。重试策略可以在消费者配置中设置，以便在处理ConsumerRecord失败时自动进行重试。
以下是在reactor-kafka中重试失败的ConsumerRecord的步骤：
- 配置重试策略：在消费者配置中，可以设置重试策略的相关参数。常见的重试策略包括固定延迟重试、指数退避重试等。可以根据实际需求选择合适的重试策略。
- 处理失败的ConsumerRecord：当消费者处理ConsumerRecord失败时，可以通过捕获异常或使用回调函数来处理失败情况。在处理失败的情况下，可以将失败的ConsumerRecord记录下来，以便后续进行重试。
- 执行重试操作：根据配置的重试策略，可以在一定的延迟后重新处理失败的ConsumerRecord。可以使用定时任务或者异步操作来执行重试操作。
- 监控重试结果：在重试过程中，可以监控每次重试的结果。如果重试成功，则可以继续处理下一个ConsumerRecord；如果重试失败，则可以根据具体情况进行进一步处理，例如记录日志或发送通知。
- 终止重试：可以设置重试的最大次数或者重试的时间窗口，当达到最大次数或者超过时间窗口后，可以终止重试操作，并根据实际情况进行处理。
在腾讯云的产品中，可以使用腾讯云消息队列 CMQ 来实现消息的可靠传输和重试。CMQ 提供了消息队列服务，可以方便地进行消息的发送和接收，并支持消息的重试和延时发送等功能。您可以通过腾讯云官网了解更多关于 CMQ 的信息：腾讯云消息队列 CMQ
另外，腾讯云还提供了云原生应用引擎 TKE，它是一种基于 Kubernetes 的容器化应用管理平台，可以帮助您快速部署和管理应用程序。您可以使用 TKE 来部署和管理使用 reactor-kafka 的应用程序，并通过 TKE 的弹性伸缩功能来应对高并发的消息处理需求。您可以通过腾讯云官网了解更多关于 TKE 的信息：腾讯云云原生应用引擎 TKE
相关·内容
我正在尝试reactor-kafka消费消息。其他一切都很好，但我想为失败的消息添加一个retry(2)。默认情况下，spring-kafka已经重试了3次失败的记录，我想用reactor-kafka来实现同样的效果。 我使用spring-kafka作为reactive-kafka的包装器。下面是我的消费者模板： reactiveKafkaConsumerTemplate
.receiveAutoAck()
.map(ConsumerRecord::value)
.flatMap(this::co
浏览 38提问于2021-10-01得票数 3
回答已采纳
我们正在用卡夫卡批量消费。我们使用X消息，将它们放在MYSQL上，然后提交它们。
有时我们会部分插入MYSQL (重复记录，其他故障等等)
使用文档中的此示例：
List<ConsumerRecord<String, String>> buffer = new ArrayList<>();
while (true) {
ConsumerRecords<String, String> records = consumer.poll(100);
for (ConsumerRecord<String, St
浏览 1提问于2018-11-26得票数 2
我使用spring的配置如下：
@Component
public class RetryableKafkaListener {
private static final Logger LOGGER = LoggerFactory.getLogger(RetryableKafkaListener.class);
@RetryableTopic(
attempts = "1",
backoff = @Backoff(delay = 1000, multiplier = 2.0),
autoCreateTopics =
浏览 4提问于2021-09-23得票数 0
2回答
我正在使用卡夫卡听众的消费者信息。我已经启用了手动确认。在成功的情况下，我使用了Acknowledgment.acknowledge()方法来提交偏移量。如果由于某些API调用失败而导致异常，我没有提交它。
我有点困惑，因为除非我重新启动应用程序，否则未提交的消息不会再次被消耗。我是卡夫卡的新手，所以我肯定我错过了一些非常基本的东西，但是搜索却是徒劳的。有谁能给我解释一下，或者指给我找个合适的来源，让我知道这里有什么问题？
@KafkaListener(topics = "temp_topic", groupId = "temp-consumer", cont
浏览 0提问于2018-05-22得票数 2
在我的侦听器中，在使用消息之后，如果发生任何异常，我就会抛出一个异常。如果它成功了，那么我承认。但是，即使抛出异常，偏移量也不会被设置回来。也就是说，重试没有按照预期发生。错误事件不再出现。
此外，我还看到我没有使用所有预期的消息。我是不是做错了什么？
@Bean
public ConcurrentKafkaListenerContainerFactory<String,Result<FormatException,String>> kafkaListenerContainerFactory(){
ConcurrentKafkaListenerContainer
浏览 2提问于2020-03-11得票数 1
1回答
我正在处理实现错误，并有一个问题。
让我解释一下这个问题：
我接收到一批消息，对for循环中的每个消息进行DB查找，然后需要收集列表中的所有查找对象，并调用bulk insert存储proc，并使用此对象列表对存储的proc进行大容量更新。
现在，让我们假设在查找过程中出现了一些异常。我想重试这条信息。在这种情况下，我尝试使用DefaultErrorHandler。但是，根据文档，当抛出带有元素索引的BatchListenerFailedException时，它会在索引之前提交记录的偏移量。但是，正如我所说的，我需要在查找之后执行批量插入和更新，这样我就不想在索引之前提交偏移量，这些记录还没有
浏览 19提问于2022-01-12得票数 0
回答已采纳
我目前正在使用Spring来使用来自主题的消息以及@Retry of Spring。因此，基本上，我正在重新尝试处理消费者的信息，以防出现错误。但是，在这样做时，我希望避免KafkaMessageListenerContainer抛出的异常消息。相反，我想显示一条自定义消息。我尝试在ConcurrentKafkaListenerContainerFactory中添加一个错误处理程序，但是在这样做时，不会调用我的重试。有人能指导我如何显示自定义异常消息以及@Retry场景吗？下面是我的代码片段：
ConcurrentKafkaListenerContainerFactory Bean Confi
浏览 14提问于2020-06-24得票数 0
我试图通过注释实现spring错误处理。
所以我创建了一个bean kafkaListenerErrorHandler。
@Bean
public KafkaListenerErrorHandler kafkaListenerErrorHandler(ErrorHandler errorHandler) {
KafkaListenerErrorHandler kafkaListenerErrorHandler = (message, exception) -> {
MessageHeaders headers = message.getHeaders();
浏览 8提问于2020-03-05得票数 0
回答已采纳
我正在使用在我的ErrorHandlingDeserialiser上发送的DefaultKafkaConsumerFactory来处理反序列化错误。
我有代码定制
try (ErrorHandlingDeserializer<MyEvent> errorHandlingDeserializer = new ErrorHandlingDeserializer<>(theRealDeserialiser)) {
errorHandlingDeserializer.setFailedDeserializationFunction(myCustomFun
浏览 2提问于2021-07-19得票数 0
回答已采纳
1回答
春云流卡夫卡粘结剂重试
、、、
我有春季云流应用与卡夫卡绑定，消费和发送信息。在应用程序中，我使用重试策略配置自定义错误处理程序，并向处理程序中添加不可还原的异常。配置示例：
@Bean
public ListenerContainerCustomizer<AbstractMessageListenerContainer<?, ?>> customizer(
SeekToCurrentErrorHandler customErrorHandler
) {
return (((container, destinationName, group) -> {
conta
浏览 6提问于2022-05-25得票数 0
回答已采纳
我计划使用spring批处理侦听器进行批处理。我正在寻找这两个场景的几个样本。
如何在批处理中实现过滤记录策略？更新:来自文档-“此外，还提供了一个FilteringBatchMessageListenerAdapter，供您使用批处理消息侦听器时使用。”还不清楚。我没有看到任何容器工厂方法来设置这个filterbatchmessagelisteneradapter对象或filter implementation.。
以下是我的批处理侦听器筛选策略代码：
@Bean
public ConcurrentKafkaListenerContainerFactory<?, ?> k
浏览 2提问于2021-02-20得票数 1
回答已采纳
我目前正在使用重试模板，并尝试用SeekToCurrentErrorHandler提供的新功能(Recoverer和BackOff)替换它。然而，我还没有得到完全相同的行为。使用重试模板，交付尝试了5次，但有回退。当重试次数耗尽时，会记录一个错误。SeekToCurrentErrrorHandler随后寻求回来，然后该过程又重新开始，以5次新的backoff重试开始。 我现在将其更改为以下内容： @Bean
public ConcurrentKafkaListenerContainerFactory<Object, Object> kafkaListenerContainer
浏览 107提问于2020-07-24得票数 0
我的多分支声明Jenkins管道在SCM签出过程中经常失败，出现超时错误，并且在一次或两次重试后才能工作。是否有自动重试SCM签出？
Jenkinsfile
agent {
label "agent1"
}
stages {
stage('Test') {
parallel {
stage('Test1') {
steps { sh 'echo Test 1 passed' }
}
stage(&
浏览 2提问于2021-01-16得票数 0
1回答
我们在我们的eco系统中使用了SpringDataMongoDB2.2.3.RELEASE和Mongo4.0。如果网络连接失败，我们需要配置重试次数。retryReads()和retryWrites()启用了这些操作，但它们将重试多少次？
有一个属性要设置超时，驱动程序将等待和重试，然后给出异常。这是否意味着它只会重试一次？
浏览 4提问于2020-06-17得票数 1
回答已采纳
我正在尝试在spring-boot中实现一个Kafka重试消费者，并使用SeekToCurrentErrorHandler进行重试。 我已将退避策略设置为有5次重试尝试。 我的问题是，假设重试的第一次尝试，异常是‘数据库不可用’，第二次尝试db是可用的，但在另一个步骤中存在另一个失败，如超时，在这种情况下，重试计数将回到零并开始新的尝试，或者将继续仅尝试第一次重试后的剩余尝试。 我的要求是每次异常与之前抛出的异常不同时，将重试计数重置为零。 如何在Kafka中实现这一点？ 这是我的消费者配置。 @Bean
public ConcurrentKafkaListenerContainer
浏览 115提问于2020-11-06得票数 0
我有一个反应式的kafka应用程序，它从一个主题读取数据，并将数据写入另一个主题。该主题有多个分区，我希望创建与该主题中的分区相同数量的消费者(在同一消费者组中)。根据我对此thread的理解，.receive()将只创建一个KafkaReceiver实例，该实例将从主题中的所有分区读取数据。所以我需要多个接收器来并行地从不同的分区读取数据。 为此，我编写了以下代码： @Bean
public ReceiverOptions<String, String> kafkaReceiverOptions(String topic, KafkaProperties kafkaPr
浏览 31提问于2021-10-27得票数 1
回答已采纳
1回答
我实现了一个自定义侦听器，即不使用@KafkaListener注释，因为我的应用程序需要动态链接到主题。我已经看到了迁移到Spring kafka 2.6.x的建议，但我不能升级，因为我被Spring 5.1.X.RELEASE卡住了(至少现在是这样)，这意味着我只能使用Spring-kafka 2.2.x。
我的问题是，如何使用Spring-kafka 2.2.x实现重试、恢复和错误处理？
ConcurrentKafkaListenerContainerFactory listenerFactory = new ConcurrentKafkaListenerContainerFactory(
浏览 31提问于2021-02-13得票数 0
回答已采纳
一次拥有2条Consumer轮询记录，即：
@Bean
ConsumerFactory<String, String> consumerFactory() {
Map<String, Object> config = Map.of(
BOOTSTRAP_SERVERS_CONFIG, "localhost:9092",
GROUP_ID_CONFIG, "my-consumers",
AUTO_OFFSET_RE
浏览 9提问于2021-12-15得票数 0
回答已采纳
我想重新发布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
我正在尝试在topicPattern订阅多个主题的消费者上实现kafka非阻塞重试。当我尝试创建一个重试RetryTopicConfiguration时，我会得到这个错误。
No topics were provided for RetryTopicConfiguration for method dynamicConsumer in class KafkaConsumerClass
这里，dynamicConsumer是用topicPattern注释的函数。
样例配置
props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, St
浏览 15提问于2021-12-06得票数 0
提前谢谢你
我的问题很简单
例如，我创建了一个工作，是发送电子邮件给用户。但是由于语法错误或任何其他异常而失败，因此laravel将在failed_jobs数据库表中将此作业视为失败的作业，例如，在2天后，我发现了代码中的问题并解决了它，现在我想再次运行失败的作业，以便完成任务。
那怎么做呢？
浏览 22提问于2020-09-18得票数 5
回答已采纳
在Spring批处理消费者中，我想跳过不可还原的异常。
如果抛出异常，我已经向DefaultErrorHandler.addNotRetryableExceptions添加了一个异常。我原以为它会去ConsumerRecordRecoverer而不会退休。
在我的测试中，它最终会转到ConsumerRecordRecoverer，但只在它被重试了n次之后。
DefaultErrorHandler.addNotRetryableExceptions在批处理模式下工作吗？也许我看错了春季卡夫卡文献。
我的示例测试代码，这是使用spring:2.9.2
@SpringBootApplication
浏览 3提问于2022-10-21得票数 1
回答已采纳
我正在为spring kafka项目实现异常处理，我有自己的DeadLetterPublishingRecoverer来处理在Kafka listener中发生的异常，整个流程是完美的，即当我在逻辑中抛出任何异常时，我可以根据框架实现将其记录并发送到DLQ主题。问题发生在消费者记录上，如果我更改了消费者记录值中的任何内容，相同的消费者记录将被发布到DLQ主题，而这些更改实际上在我的senairo中是错误的，我希望记录原始消息。 ..。 @KafkaListener(topics = "${message.topic.name}",containerFactory
浏览 14提问于2020-09-15得票数 0
1回答
我正在做一个有两个服务的项目:阅读，转换信息，然后写到另一个卡夫卡。这两种服务的Kafka配置是不同的。这是我的application.yml
spring:
kafka:
bootstrap-servers: localhost:9092
sourcetopic1: topic1
destinationtopic1 : topic2
sourcetopic2: topic3
destinationtopic2 : topic4
producer:
key-serializer: org.apache.kafka.comm
浏览 1提问于2021-09-10得票数 1
回答已采纳
1回答
如果我的lambda函数失败，那么AWS中是否有任何方法在3-4小时后调用相同的函数。
如果是的话，这样做的流程是什么？
浏览 5提问于2022-05-30得票数 0
我是Spring批处理的新手，使用spring引导，我阅读了几乎所有可用的文章，但没有找到用于用例的确切场景/片段，如下所示。
从DB读取数据(例如:5项)
处理数据:如果API调用失败，则调用API和，跳过项并处理下一个项。(对所有5个项目按顺序排列)
所有跳过的项都应记录在db中。
在处理所有5项失败/成功方案后，然后在一段时间间隔后重试失败项。
我已经使用了RetryTemplate，但是在那里它会在异常情况下立即重新尝试这个项目，所以它不适合我。
我创建了跳过策略，它跳过了异常，给了我预期的行为，但无法开始重试尝试(3)以同样的方式。
有人能帮我举几个例子并指导
浏览 0提问于2018-10-29得票数 0
2回答
我有一个Springboot应用程序配置了Springboot，在这里我想处理在听主题时可能发生的各种错误。如果由于反序列化或任何其他异常而丢失/无法使用任何消息，则将进行2次重试，然后将消息记录到错误文件中。我有两种方法可以遵循:-
DeadLetterPublishingRecoverer):-优先方法(使用SeekToCurrentErrorHandler与SeekToCurrentErrorHandler)
@Autowired
KafkaTemplate<String,Object> template;
@Bean(name = "kafkaSourceProv
浏览 11提问于2020-08-03得票数 0
回答已采纳
上下文：
我正在使用spring重试restTemplate调用。
restTemplate调用是从卡夫卡侦听器调用的。kafka侦听器还被配置为重新尝试错误(如果在进程中抛出任何异常，而不仅仅是restTemplate调用)。
目标：
当错误来自已耗尽的重试模板时，我想阻止kafka重新尝试。
实际行为:
当retryTemplate耗尽所有重试时，将引发原始异常。这样，我就无法识别错误是否被retryTemplate重试了。
期望行为：
当retryTemplate耗尽所有重试时，将原始异常包装在RetryExhaustedException中，这将允许我从卡夫卡重试中将其列入黑名单。
问
浏览 7提问于2019-10-17得票数 1
回答已采纳
Kafka v2.4消费者配置：-
kafka.consumer.auto.offset.reset=earliest
kafka.consumer.auto.commit=false
Kafka消费容器配置：-
@Bean
public ConcurrentKafkaListenerContainerFactory<String, PayoutDto> kafkaPayoutStatusPoolListenerContainerFactory() {
ConcurrentKafkaListenerContainerFactory<String, PayoutDto&
浏览 2提问于2021-04-13得票数 1
回答已采纳
retries参数的状态
设置大于零的值将导致客户端重新发送发送失败而可能出现短暂错误的任何记录。请注意，此重试与客户端在收到错误时重新启动记录没有什么不同。
我要确认的是由kafka框架自动进行的重试，或者需要从客户端进行任何额外的处理。
如果它是自动完成的，并且说retries被设置为1。如果record的发送在第二次尝试中也失败了，那么卡夫卡是否会以不同的方式告诉我适当的错误信息:何时重试，何时停止重试。
浏览 2提问于2017-08-06得票数 1
我有一个PowerShell脚本，它可以从错误队列中重放MSMQ消息。我希望NServiceBus.FLRetries能够以更多的失败进行更新。然而，情况并非如此。
我读过，但无法完全理解这个特定的标题。
我想看到的是，随着消息继续无法处理(例如，web服务不可用)，重试次数会增加。
我用的是NSB 5.2。
任何想法，我可以如何建模，如果上面的标题是不可用的。
浏览 3提问于2017-03-20得票数 0
如何在失败(失败)后重新启动测试？
我测试了ci_events_page.py (吡喃-硒测试)
yml设置(..gitlab ci.yml)：
ci_events_page.py:
stage: test
type: test
script:
- xvfb-run --server-args="-screen 0 1920x1080x24" pytest --driver Chrome ci/ci_events_page.py -v
测试没问题！但是当测试失败的时候-作业也失败了.我想在失败的时候重新启动几次测试。拜托救救我。
浏览 2提问于2018-02-06得票数 1
回答已采纳
我使用spring kafka (KafkaTemplate)来发送字符串消息。但是为了使它与旧代码兼容，我需要在消息中附加额外的CorrelationId。所以我创建了ProducerRecord对象，将我的消息作为它的值，并在它的HeaderRecord中设置CorrelationId。
producerRecord = new ProducerRecord<>(
kafkaTemplate_.getDefaultTopic(),
null,
null,
null,
浏览 6提问于2017-12-12得票数 0
回答已采纳
我使用火花流读取kafka数据，并处理每一行。
我使用下面创建一个流：
lines = KafkaUtils.createDirectStream(
jssc,
LocationStrategies.PreferConsistent(),
ConsumerStrategies.<String, String>Subscribe(topics,kafkaParams)
);
然后我使用这个代码来处理kafka的数据。
lines.foreachRDD((JavaRDD<ConsumerRec
浏览 2提问于2017-12-28得票数 3
回答已采纳
2回答
在下面的场景中，我们有一个ForEach活动在Azure Data Factory管道中运行，以将数据从源复制到目的地。最后一次CopyActivity花了4:10:33，但ForEach活动在36分钟后宣布成功: 4:46:12。问题是，为什么ForEach活动需要这36分钟的额外时间？在宣布成功或失败之前，ForEach还需要巩固子活动的结果吗？
浏览 2提问于2021-03-01得票数 1
回答已采纳
我有几个关于AWS lambdas的问题，我在文档中找不到太多细节
如何在AWS Lambda中增加重试次数？
如果重试次数达到最大值，而整个Lambda都失败了，我如何才能收到某种通知？
浏览 0提问于2019-03-09得票数 0
2回答
问题
我有一个云函数，它根据文档state字段中的更改更新Cloud文档。这些更新必须以特定的顺序进行。当快速连续进行两个更改时，无法保证云函数将以正确的顺序运行。是否有一种方法可以让CloudFi还原事务重试直到它成功或超时？
Document.state设置为stage1
Document.state被更新为stage2
Document.state被更新为stage3
云函数被触发并读取stage3
云函数被触发并读取stage2
在Cloud文档中，讨论了在失败时重试事务的能力。但是，此选项在GCP控制台的部分(未显示在Firebase控制台中)中显示为灰色
浏览 0提问于2018-02-19得票数 2
在我的C#程序中，我使用实体框架将本地Server数据库与QuickBooks数据同步。从QuickBooks获取数据似乎没有任何问题。但是，在对实体进行批量提交时，我遇到了一个绊脚石。
目前，我正在使用可配置的实体数构建DataContext，然后批量提交实体。到目前为止，该批还没有失败，但如果它失败了呢？与此作斗争的想法是迭代批处理并一次提交每个实体，然后记录导致提交失败的实体。
但是，我看不到在数据上下文中这样做的方法，因为在使用SaveChanges()时，这似乎是一个全部或根本不重要的问题。是否有一种方法来处理我想要完成的事情，或者我是否应该以一种完全不同的方式来处理这些失败？
下面
浏览 3提问于2017-05-30得票数 1
回答已采纳
1回答
我有一个spring引导kafka消费者，它使用来自某个主题的数据，并将其存储在数据库中，并在存储后确认它。它运行良好，但如果应用程序在使用记录后未能获得DB连接，则会发生问题，在本例中，我们没有发送确认，但直到或除非我们更改组id并重新启动使用者，否则消息从未被消耗过。
我的消费者如下所示
@KafkaListener(id = "${group.id}", topics = {"${kafka.edi.topic}"})
public void onMessage(ConsumerRecord record, Acknowledgment ackno
浏览 0提问于2019-03-25得票数 4
我尝试使用弹簧重试来进行电路断开并重试，如下所示，它按预期工作，但问题是无法将"maxAttempts/openTimeout/resetTimeout“配置为环境变量(错误是应该是常量)。我的问题是如何使用resilience4j来实现以下要求？ 另外，请建议有一种方法可以将环境变量传递给"maxAttempts/openTimeout/resetTimeout“。 @CircuitBreaker(value = {
MongoServerException.class,
MongoSocketException.class,
浏览 13提问于2019-03-22得票数 0
回答已采纳
技术：
2.2.7enable.auto.commit = falseAckMode.BATCHisSyncCommits =false分区数:24用户数: 12 (按春季术语计算的并发性)
问题场景：夫妇kafka集群节点崩溃，kafka开始报告异常，而不是接受我们的提交。默认异步回调只是将这些错误打印到日志中。投票仍在继续。消息处理得很好(但没有提交)。10分钟后，prod支持在日志中发现错误。卡夫卡集群恢复了，苦难进程被反弹。由于提交被代理删除了一段时间，该进程处理了大约200万条重复消息。
问题：
我们可以在并发映射中保留每个TopicPartition的最新成功偏移量。因此，在代理再
浏览 3提问于2021-03-07得票数 0
在Jenkins CI中，有一个“失败后重试构建”的插件，我们可以指定在作业失败时希望构建自动触发的次数。
在Azure Devops中做了类似的事情吗，在构建失败时管道将自动触发？
浏览 2提问于2021-04-02得票数 2
我使用kafka-python来处理kafka集群中的消息：
消费者=KafkaConsumer(‘会话’，auto_offset_reset='earliest']
而True：
dict = consumer.poll(500)
for d in dict:
print d.topic, d.partition, d.value
这将给出错误“值：‘AttributeError’对象没有'TopicPartition‘属性”。
"dict“是这样的(来自”print dict“)
{TopicPartition(topic=u
浏览 2提问于2018-12-30得票数 0
我对Kafka非常陌生。我正在试图弄清楚和理解错误场景将如何为@Listener batch consumer factory工作。
我在做的是..。
我正在使用topic在batch进程中的记录，并将它们插入到DB中，如下所示.
@KafkaListener( topics = "KAFKA.TEST")
public Boolean listen(List<ConsumerRecord<String, User>> list) throws Exception {
Boolean result = null;
List<Use
浏览 4提问于2017-02-02得票数 4
我正在尝试将我的消费者配置为使用指数退避，其中消息将被处理固定次数的尝试，其中应用退避期限。但是我没有得到预期的行为。 这是我的Java代码： @EnableBinding({
MessagingConfiguration.EventTopic.class
})
public class MessagingConfiguration {
public interface EventTopic {
String INPUT = "events-channel";
@Input(INPUT)
@Nonnu
浏览 376提问于2021-04-08得票数 0
回答已采纳
我正在使用spring 2.2.8，并试图创建一个批量消费者。这里我有个问题。
假设我根据我的用户设置在每次投票中接收到大约100条记录，而在这100条记录中，如果只有一条记录因为一些下游系统错误而失败(可能是一个短暂的错误，比如DB停机了几秒钟并返回等等)，那么我是否可以为所有100条记录提交偏移量，然后将失败的记录发送到另一个主题或其他持久性存储，以便稍后处理它？
浏览 17提问于2020-09-09得票数 2
我想将不同类型的数据存储在Couchbase中的同一个桶中，在googled上搜索它的建议doc_type和类型参数，但是我不知道如何在数据中实现这个数据，我尝试了下面的代码，但我不知道如何实现？
我没有看到任何例子，任何帮助都会很感激，
JsonObject doc = JsonObject.fromJson(consumerRecord.value().toString());
String id=consumerRecord.key().toString();
bucket.upsert(JsonDocument.create(id,
浏览 0提问于2019-01-29得票数 2
回答已采纳
假设我有一个工作流操作，我需要从hdfs获取和使用一个文件(如果它存在)，否则不会失败。有没有办法在工作流配置中做到这一点？
所以，如果我有这样的东西：
<action name="run">
<java>
<main-class>AJavaProgram</main-class>
...
<arg>-files</arg>
<arg>properties.json</arg>
...
浏览 2提问于2017-07-25得票数 0
扫码
添加站长 进交流群
领取专属 10元无门槛券
手把手带您无忧上云
相关资讯
热门标签
云服务器
ICP备案
对象存储
实时音视频
云直播活动推荐
腾讯云开发者
