I'm trying to use spring validation annotation on @KafkaListener method. I was trying the following:

@Component
@Validated
public class Consumer {
    @KafkaListener(topics = "${app.kafka.consumer.topic}", groupId = "${spring.kafka.consumer.group-id}")
    public String consume(@Payload @Valid @NotNull RequestObj request,
                          @Header(KafkaHeaders.RECEIVED_TOPIC) @Valid @NotEmpty String topic) {
        ...
    }
}

And it doesn't seem to work, meaning the @NotEmpty annotation is not checked.

I also configured the configuration mentioned in the docs:

@Configuration
@EnableKafka
public class Config implements KafkaListenerConfigurer {

    @Autowired
    private LocalValidatorFactoryBean validator;
    ...

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
      registrar.setValidator(this.validator);
    }
}

But it only effect the payload part.

Is what I am trying to achieve is actually supported?

Thanks for your help.

CodePudding user response：

Built-in validation is only supported for payloads.

However, @Header has a required property, which is true by default so a null value would be rejected.

However, RECEIVED_TOPIC will never be null.