kafka-consumer-api 相关问题

用于与Apache Kafka使用者API相关的问题

负载测试kafka消费者

（我正在编辑问题，因为我认为它不够清楚）如何对我的 kafka 消费者进行负载测试？我看过很多关于负载测试 apache kafka 的文章，但没有一篇关于负载测试 con...

spring-boot jmeter kafka-consumer-api

回答 2 投票 0

Spring-kafka：打印RecordHeaders的正确方法是什么？

刚刚检查了日志，发现 ConsumerRecord 标头以绝对不可读的方式打印，原因是 RecordHeader 将值保存为 byte[]，结果在日志中我看到了

spring apache-kafka kafka-consumer-api spring-kafka

回答 1 投票 0

在 Spring Boot 应用程序中实现反应式 Kafka 监听器

我正在尝试在 Spring boot 应用程序中实现反应式 kafka 消费者，我正在查看这些示例： https://github.com/reactor/reactor-kafka/blob/master/reactor-kafka-samples/src/main/...

java kafka-consumer-api spring-kafka reactive-kafka reactor-kafka

回答 1 投票 0

使用offsets_for_times从时间戳消费

尝试使用 confluence_kafka.AvroConsumer 来消费给定时间戳的消息。如果标志： # 创建一个列表 topic_partitions_to_search = 列表( 映射（lambda p：TopicPartition（'

python apache-kafka kafka-consumer-api confluent-kafka-python

回答 2 投票 0

如何消费kafka消息上的最后一条消息或者根据时间戳消费消息？

def kafkaa（自我，auto_offset_reset，超时= 500）： group_name = "群组名称" config = {“bootstrap.servers”：“服务器”， “是...

python apache-kafka kafka-consumer-api confluent-kafka-python

回答 1 投票 0

Spring Kafka 不可重试异常过滤器

我正在尝试实现动态 Kafka 异常处理，特别是在某些条件下重试，但找不到任何方法来过滤不应该仅由其类重试的异常。我的...

java spring kafka-consumer-api spring-kafka

回答 1 投票 0

将整批发送到 dlt，无需重试

我正在使用spring kafka，并且我有一个用java spring boot编写的kafka消费者。我的消费者按批次消费，相关配置 bean 如下所示。 @豆公共消费者工厂我正在使用 spring kafka，并且我有一个用 java spring boot 编写的 kafka 消费者。我的消费者批量消费和相关配置bean如下。 @Bean public ConsumerFactory<String, Object> consumerFactory() { Map<String, Object> config = new HashMap<>(); // default configs like bootstrap servers, key and value deserializers are here config.put(ConsumerConfig.MAX_POLL_RECORDS_CONFIG, "5"); return new DefaultKafkaConsumerFactory<>(config); } @Bean public ConcurrentKafkaListenerContainerFactory<String, Object> kafkaListenerContainerFactory() { ConcurrentKafkaListenerContainerFactory<String, Object> factory = new ConcurrentKafkaListenerContainerFactory<>(); factory.setConsumerFactory(consumerFactory()); factory.getContainerProperties().setCommitLogLevel(LogIfLevelEnabled.Level.DEBUG); factory.setBatchListener(true); return factory; } 我使用消息并将这些消息发送到 API 端点。如果 api 不可用或者其余模板抛出错误，我想将整个批次发送到 DLT 而无需重试。我想要做的是将整个批次发送到 DLT，而不重试。如果我们抛出 BatchListenerFailedException，则批次中拥有特定索引号的消息将发送到 DLT。在 BatchListenerFailedException 中，我们只能传递一个整数值作为索引值，而不是一个列表。但我想要的是将整个批次按原样发送到 DLT 主题，而无需重试。有办法实现吗？我的Spring Kafka版本是2.8.6 编辑我的默认错误处理程序如下 @Bean public CommonErrorHandler commonErrorHandler() { ExponentialBackOffWithMaxRetries exponentialBackOffWithMaxRetries = new ExponentialBackOffWithMaxRetries(5); exponentialBackOffWithMaxRetries.setInitialInterval(my val); exponentialBackOffWithMaxRetries.setMultiplier(my val); exponentialBackOffWithMaxRetries.setMaxInterval(my val); DefaultErrorHandler errorHandler = new DefaultErrorHandler( new DeadLetterPublishingRecoverer(kafkaTemplate(), (record, exception) -> new TopicPartition(record.topic() + "-dlt", record.partition())), exponentialBackOffWithMaxRetries); errorHandler.addNotRetryableExceptions(ParseException.class); errorHandler.addNotRetryableExceptions(EventHubNonRetryableException.class); return errorHandler; } 在我的例子中使用ExponentialBackOffWithMaxRetries而不是FixedBackOff。就我而言，我有 3 个场景。 1 - 重试消息并将其发送到 DLT（抛出除 BatchListenerFailedException 之外的任何其他异常） 2 - 将批次中的几条消息发送到 DLT，无需重试（为此使用 BatchListenerFailedException） 3 - 将整个批次发送到 DLT，无需重试。第三个是我苦苦挣扎的地方。如果我发送一些其他异常，那么它会重试几次。（即使我用FixedBackOff代替ExponentialBackOffWithMaxRetries）扔BatchListenerFailedException以外的其他东西；将 DefaultErrorHandler 与 DeadLetterPublishingRecoverer 一起使用，无需重试 (new FixedBackOff(0L, 0L))。编辑从版本3.0.0、2.9.3、2.8.11开始，您可以为批量错误配置不可重试的异常。 https://github.com/spring-projects/spring-kafka/issues/2459 看 /** * Add exception types to the default list. By default, the following exceptions will * not be retried: * <ul> * <li>{@link DeserializationException}</li> * <li>{@link MessageConversionException}</li> * <li>{@link ConversionException}</li> * <li>{@link MethodArgumentResolutionException}</li> * <li>{@link NoSuchMethodException}</li> * <li>{@link ClassCastException}</li> * </ul> * All others will be retried, unless {@link #defaultFalse()} has been called. * @param exceptionTypes the exception types. * @see #removeClassification(Class) * @see #setClassifications(Map, boolean) */ @SafeVarargs @SuppressWarnings("varargs") public final void addNotRetryableExceptions(Class<? extends Exception>... exceptionTypes) { add(false, exceptionTypes); notRetryable(Arrays.stream(exceptionTypes)); } 请注意，2.8.x 现已不再支持 OSS。 https://spring.io/projects/spring-kafka#support 您能就第二个场景的实施与我联系吗？这将是一个很大的帮助。谢谢

apache-kafka kafka-consumer-api spring-kafka

回答 2 投票 0

如何将LinkedHashMap转换为自定义对象？

我的 Kafka Consumer 端有 2 个主题：主题 A 和主题 B。两者都采用相同的方法。详细方法如下； @Slf4j @服务公共类 KafkaConsumerService { @KafkaListe...

java spring-boot apache-kafka kafka-consumer-api

回答 1 投票 0

如何将LinkedHasMap转换为自定义对象？

我的 Kafka Consumer 端有 2 个主题。主题 A 和主题 B。两者都采用相同的方法。详细方法如下； @Slf4j @服务公共类 KafkaConsumerService { @KafkaListe...

java spring-boot apache-kafka kafka-consumer-api

回答 1 投票 0

Kafka Consumer 无法使用 KafkaJS 在本地工作

我正在尝试使用 KafkaJS 在本地运行 Kafka 消费者。它表明消费者正在运行，但它没有从生产者主动向其中推送事件的主题进行消费。消费者