Kafka ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG = false不起作用

bq9c1y66  于 2023-06-21  发布在  Apache
关注(0)|答案(1)|浏览(86)

SpringBoot2 SpringKafka
我现在遇到一个问题，ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG = false不能正常工作。我已经将allow. auto. create. topics设置为false，但是Topic仍然会在运行时自动创建。
下面是我所做的代码更改。
EmbeddedKafkaIntegrationTest.java (Unit Test class)

package com.mbag.kafka;

import org.apache.kafka.clients.admin.AdminClientConfig;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
import org.junit.jupiter.api.Test;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.boot.test.context.SpringBootTest;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Import;
import org.springframework.kafka.annotation.EnableKafka;
import org.springframework.kafka.config.ConcurrentKafkaListenerContainerFactory;
import org.springframework.kafka.core.ConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaConsumerFactory;
import org.springframework.kafka.core.DefaultKafkaProducerFactory;
import org.springframework.kafka.core.KafkaAdmin;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.core.ProducerFactory;
import org.springframework.kafka.test.context.EmbeddedKafka;
import org.springframework.test.annotation.DirtiesContext;
import org.springframework.test.context.ContextConfiguration;
import org.springframework.util.StringUtils;

import java.time.Duration;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Properties;
import java.util.concurrent.TimeUnit;

import static org.junit.jupiter.api.Assertions.assertTrue;

@SpringBootTest(classes = {KafkaConsumerService.class, KafkaProducerService.class})
@Import(com.mbag.kafka.EmbeddedKafkaIntegrationTest.KafkaConfigurationTest.class)
@DirtiesContext
@EmbeddedKafka(partitions = 1, brokerProperties = {
        "listeners=PLAINTEXT://localhost:9092",
        "port=9092",
        "auto.create.topics.enable=true"
})
public class EmbeddedKafkaIntegrationTest {

    @Autowired
    private KafkaConsumerService consumer;

    @Autowired
    private KafkaProducerService producer;

    @Value("${test.topic}")
    private String topic;

    @Test
    public void givenEmbeddedKafkaBroker_whenSendingWithSimpleProducer_thenMessageReceived() throws Exception {
        String data = "Sending with our own simple KafkaProducer";

        producer.send(topic, data);

        boolean messageConsumed = consumer.getLatch().await(3, TimeUnit.SECONDS);

        assertTrue(messageConsumed);
        assertTrue(consumer.getPayload().contains(data));
    }

    @Configuration
    @EnableKafka
    static class KafkaConfigurationTest {

        @Bean
        public KafkaAdmin admin() {
            Map<String, Object> configs = new HashMap<>();
            configs.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            return new KafkaAdmin(configs);
        }

        @Bean("kafkaListenerContainerFactory")
        ConcurrentKafkaListenerContainerFactory<Integer, String> kafkaListenerContainerFactory() {
            ConcurrentKafkaListenerContainerFactory<Integer, String> factory = new ConcurrentKafkaListenerContainerFactory<>();
            factory.setConsumerFactory(consumerFactory());
            factory.setRecordFilterStrategy(record -> record.value().contains("test"));
            return factory;
        }

        @Bean
        public ConsumerFactory<Integer, String> consumerFactory() {
            return new DefaultKafkaConsumerFactory<>(consumerConfigs());
        }

        @Bean
        public Map<String, Object> consumerConfigs() {
            Map<String, Object> props = new HashMap<>();
            props.put(ConsumerConfig.ALLOW_AUTO_CREATE_TOPICS_CONFIG, "false");
            props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest"); // earliest latest
            props.put(ConsumerConfig.GROUP_ID_CONFIG, "localtest");
            props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, StringDeserializer.class);
            return props;
        }

        @Bean
        public ProducerFactory<String, String> producerFactory() {
            Map<String, Object> configProps = new HashMap<>();
            configProps.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
            configProps.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            configProps.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
            return new DefaultKafkaProducerFactory<>(configProps);
        }

        @Bean
        public KafkaTemplate<String, String> kafkaTemplate() {
            return new KafkaTemplate<>(producerFactory());
        }

    }

}

KafkaConsumerService.java

package com.mbag.kafka;

import lombok.Data;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.RetryableTopic;
import org.springframework.kafka.retrytopic.FixedDelayStrategy;
import org.springframework.retry.annotation.Backoff;
import org.springframework.stereotype.Component;

import java.util.concurrent.CountDownLatch;

@Component
@Data
public class KafkaConsumerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaConsumer.class);

    private CountDownLatch latch = new CountDownLatch(1);
    private String payload;

    @KafkaListener(topics = "${test.topic}", groupId = "localtest", containerFactory = "kafkaListenerContainerFactory")
    public void receive(ConsumerRecord<?, ?> consumerRecord) {
        LOGGER.info("Received payload='{}'", consumerRecord.toString());
        payload = consumerRecord.toString();
        latch.countDown();
    }

    public void resetLatch() {
        latch = new CountDownLatch(1);
    }

}

KafkaProducerService.java

package com.mbag.kafka;

import org.apache.kafka.clients.producer.KafkaProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

@Component
public class KafkaProducerService {

    private static final Logger LOGGER = LoggerFactory.getLogger(KafkaProducer.class);

    @Autowired
    private KafkaTemplate<String, String> kafkaTemplate;

    public void send(String topic, String payload) {
        LOGGER.info("sending payload='{}' to topic='{}'", payload, topic);
        kafkaTemplate.send(topic, payload);
    }

}

application.yml

spring:
  kafka:
    consumer:
      auto-offset-reset: earliest
      group-id: localtest
      properties:
        allow.auto.create.topics: false
test:
  topic: embedded-test-topic
  topic-test: testing-only

基于上面的代码更改，我通过更改代理和消费者之间的标志进行了一些测试。
1.如果我在代理级别更改了auto.create.topics.enable = false，测试用例将由于无法订阅主题而失败。这很好，因为预期的行为。
1.如果我启用了auto. create. topics. enable = true，但禁用了allow. auto. create. topics = false，主题仍然是自动创建的。这不是意外行为。
我已经在控制台和调试模式下检查了日志，该属性确实设置为false。但事情并没有像预期的那样发展。
上面的测试使用了@EmbeddedKafka。我试过使用TestContainers Kafka，结果还是一样。不工作。
有人能告诉我，如果我在这里做了任何不正确的修改吗？

    • 更新：**经过测试，主题似乎是在Producer部分创建的。我为consume创建了一个简单的测试用例（没有produce代码），主题没有被创建。我为produce创建了另一个简单的测试用例（没有消费代码），我看到主题被自动创建。

有谁知道我如何禁用主题自动创建生产者？或者这是否是预期功能？
我理解我可以在代理级别禁用主题自动创建（auto. create. topics. enable = false）。然而，有时我们可能不允许在底层修改属性。

apache-kafka

1条答案

按热度按时间
r3i60tvu

r3i60tvu1#

我如何禁用主题自动创建生产者？
您将禁用代理上的自动创建。
理想情况下，你也不允许从消费者。否则，您将创建没有活动生产者的空主题（例如，您错误地输入了主题名称，并且您将不会收到任何消息或警告，因为主题将被创建为空）。
在其他测试运行之前，您可以自己使用AdminClient来实际创建主题。

赞(0）分享  回复(0）举报  2023-06-21
我来回答

相关问题

    查看更多

    热门标签

    更多
    JavaquerypythonNode开发语言requestUtil数据库Table后端算法LoggerMessageElementParser

    最新问答

    更多
    • 蜀ICP备2022004421号-2 NULL123 https://www.null123.com © All rights reserved
    • 本站内容来源互联网,如果侵犯您的权益请联系我们删除, 联系方式：448109455@qq.com