Validating Spring Kafka payloads

Question

I am trying to set up a service that has both a REST (POST) endpoint and a Kafka endpoint, both of which should take a JSON representation of the request object (let's call it Foo). I would want to make sure that the Foo object is valid (via JSR-303 or whatever). So Foo might look like:

public class Foo {
    @Max(10)
    private int bar;

    // Getter and setter boilerplate
}

Setting up the REST endpoint is easy:

@PostMapping(value = "/", produces = MediaType.APPLICATION_JSON_VALUE)
public ResponseEntity<String> restEndpoint(@Valid @RequestBody Foo foo) {
    // Do stuff here
}

and if I POST, { "bar": 9 } it processes the request, but if I post: { "bar": 99 } I get a BAD REQUEST. All good so far!

The Kafka endpoint is easy to create (along with adding a StringJsonMessageConverter() to my KafkaListenerContainerFactory so that I get JSON->Object conversion:

@KafkaListener(topics = "fooTopic")
public void kafkaEndpoint(@Valid @Payload Foo foo) {
    // I shouldn't get here with an invalid object!!!
    logger.debug("Successfully processed the object" + foo);

    // But just to make sure, let's see if hand-validating it works
    Validator validator = localValidatorFactoryBean.getValidator();
    Set<ConstraintViolation<SlackMessage>> errors = validator.validate(foo);
    if (errors.size() > 0) {
        logger.debug("But there were validation errors!" + errors);
    }
}

But no matter what I try, I can still pass invalid requests in and they process without error.

I've tried both @Valid and @Validated. I've tried adding a MethodValidationPostProcessor bean. I've tried adding a Validator to the KafkaListenerEndpointRegistrar (a la the EnableKafka javadoc):

@Configuration
public class MiscellaneousConfiguration implements KafkaListenerConfigurer {
    private Logger logger = LoggerFactory.getLogger(this.getClass());

    @Autowired
    LocalValidatorFactoryBean validatorFactory;

    @Override
    public void configureKafkaListeners(KafkaListenerEndpointRegistrar registrar) {
        logger.debug("Configuring " + registrar);
        registrar.setMessageHandlerMethodFactory(kafkaHandlerMethodFactory());

    }

    @Bean
    public MessageHandlerMethodFactory kafkaHandlerMethodFactory() {
        DefaultMessageHandlerMethodFactory factory = new DefaultMessageHandlerMethodFactory();
        factory.setValidator(validatorFactory);
        return factory;
    }
}

I've now spent a few days on this, and I'm running out of other ideas. Is this even possible (without writing validation into every one of my kakfa endpoints)?

Answer 1

Sorry for the delay; we are at SpringOne Platform this week.

The infrastructure currently does not pass a Validator into the payload argument resolver. Please open an issue on GitHub.