我正在尝试测试在@kafkalistener使用的containerlistenerfactory中定义的错误处理程序。
我有不同的重试次数，因为我想测试的侦听器抛出的异常是正确的。但是在第一次抛出异常并转到错误处理程序之后，我得到了一个非法的状态异常，因此它不会在我编写的测试中重试。同样的代码在实际的设置中工作。
这是我得到的一个例外。
异常后寻求电流；嵌套的异常是org.springframework.kafka.listener.listenerexecutionfailedexception
我希望测试重试10次，然后在恢复中打印消息。但它不会重试，因为errorhandler正在抛出非法的状态异常。
有人能建议一下吗？

@Configuration
@EnableKafka
public  class Config {

    public static boolean seekPerformed;

    public static int retries;

    private Integer retryCount=10;

    private Integer RetryCount2=5;

    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;

    @Spy
    private errorCodes errorCodes;

    @Bean
    public ConcurrentKafkaListenerContainerFactory<String, Anky>kafkaListenerContainerFactory(EmbeddedKafkaBroker embeddedKafka) {
        ConcurrentKafkaListenerContainerFactory<String, Anky>factory =
                new ConcurrentKafkaListenerContainerFactory<>();
        factory.setConsumerFactory(consumerFactory(embeddedKafka));
        factory.getContainerProperties().setDeliveryAttemptHeader(true);
        factory.getContainerProperties().setAckMode(AckMode.MANUAL_IMMEDIATE);
        SeekToCurrentErrorHandler errorHandler = new SeekToCurrentErrorHandler((record, exception) -> {
            System.out.println(
                    "RetryPolicy**limit has been exceeded! You should really handle this better." + record.key());
        });
        errorHandler.setBackOffFunction((record, exception) -> {
            retries++;
            seekPerformed = true;
            int maxRetryCount = retryCount+ retryCount2;
            Anky msg = (Anky) record.value();

            if (msg.getErrorCode.equals(getExceptionA())) {
                return new FixedBackOff(0L,Long.valueOf(retryCount));
            }
            else {
                return new FixedBackOff(0L,Long.valueOf(retryCount2));
            }

        });
        errorHandler.setCommitRecovered(true);
        factory.setErrorHandler(errorHandler);
        factory.setConcurrency(2);
        //errorHandler.setLogLevel(Level.INFO);
        factory.setStatefulRetry(true);
        return factory;
    }

    @Bean
    public DefaultKafkaConsumerFactory<String, Anky> consumerFactory(EmbeddedKafkaBroker embeddedKafka) {
        //return new DefaultKafkaConsumerFactory<>(consumerConfigs(embeddedKafka));
        return new DefaultKafkaConsumerFactory<>(consumerConfigs(embeddedKafka), new StringDeserializer(),
                new JsonDeserializer<>(EdealsMessage.class, false));
    }

    @Bean
    public Map<String, Object> consumerConfigs(EmbeddedKafkaBroker embeddedKafka) {
        Map<String, Object> props = new HashMap<>();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokerAddress(0).toString());
        props.put(ConsumerConfig.GROUP_ID_CONFIG, "retry-grp");
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, false);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
        //props.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, new StringDeserializer());
        //props.put(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, new JsonDeserializer<>(EdealsMessage.class, false));
        return props;
    }

    @Bean
    public ProducerFactory<String, Object> testProducerFactory(EmbeddedKafkaBroker embeddedKafka) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, embeddedKafka.getBrokerAddress(0).toString());
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, JsonSerializer.class);
        return new DefaultKafkaProducerFactory<>(configs);
    }

    @Bean
    public KafkaTemplate<String, Object> testKafkaTemplate(EmbeddedKafkaBroker embeddedKafka) {
        KafkaTemplate<String, Object> kafkaTemplate = new KafkaTemplate<>(testProducerFactory(embeddedKafka));
        kafkaTemplate.setDefaultTopic("sr1");
        return kafkaTemplate;
    }

    @KafkaListener(topics = "sr1", groupId = "retry-grp",containerFactory = "kafkaListenerContainerFactory")
    public void listen1(ConsumerRecord<String, Anky> record,
            @Header(KafkaHeaders.DELIVERY_ATTEMPT) int delivery) throws AppException{

        try {
            throw new AppException(//this is our custom exception in the application);
        }
        catch(AppException se) {
            if(record.value().getNewErrorCode().equals(se.getErrorCode())) {
                System.out.println("are you here?");
                throw se;
            }
    }

}

@EnableKafka
@SpringBootTest
@ExtendWith(SpringExtension.class)
@ContextConfiguration(classes = Config.class)
@EmbeddedKafka(
        partitions = 1, 
        controlledShutdown = true, topics = {"sr1"},
        brokerProperties = {
            "listeners=PLAINTEXT://localhost:3333", 
            "port=3333"
    })
public class KafkaRetryTest {

    @Autowired
    private Config config;

    @Autowired
    private KafkaTemplate<String, Object> template;

    @Autowired
    KafkaListenerEndpointRegistry kafkaListenerEndpointRegistry;

    @Autowired
    EmbeddedKafkaBroker kafkaEmbedded;

    @Spy
    private ErrorCodes errorCodes;

    @BeforeEach
    public void setUp() throws Exception,SystemException {
      for (MessageListenerContainer messageListenerContainer : kafkaListenerEndpointRegistry.getListenerContainers()) {
        ContainerTestUtils.waitForAssignment(messageListenerContainer, 
        kafkaEmbedded.getPartitionsPerTopic()
        );
      }
    }

    @Test
    public void testStatefulRetry() throws Exception {
        Anky msg = new Anky();
        msg.setNewErrorCode(errorCodes.getExceptionA());
        this.template.send("sr1","3323800",msg);
        assertThat(this.config.seekPerformed).isTrue();
        System.out.println("******"+this.config.retries);
}   
}

*******更新的代码，我无法在embeddedkafka上使用bootstrapserversproperty=“spring.kafka.bootstrap-servers，因为它一直试图转到localhost:9092 and 超时

这是我的测试文件配置

@EnableKafka
@SpringBootTest(classes=MyConsumer.class)
@ExtendWith(SpringExtension.class)
@DirtiesContext
@ContextConfiguration(classes = AppConfig.class)
@EmbeddedKafka(
    partitions = 1, 
     topics = {"test_topic"}
             , bootstrapServersProperty ="spring.kafka.bootstrap-servers"
    )
public class KafkaConsumerTest {
}

appconfig***（bootstrapaddress从属性文件读取）

@Value(value = "${kafka.bootstrapAddress}")
    private String bootstrapAddress;

@Bean
    public ConsumerFactory<String, Anky> consumerFactory() {

        Map<String, Object> props = kafkaProperties.buildConsumerProperties();
        props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapAddress);
        props.put(ConsumerConfig.GROUP_ID_CONFIG, groupId);
        props.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, autoCommit);
        props.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, autoOffsetReset);

        return new DefaultKafkaConsumerFactory<>(props, new StringDeserializer(),
                new JsonDeserializer<>(Anky.class, false));
    }