验证 Spring Kafka 有效负载

Question

我正在尝试设置一个同时具有 REST (POST) 端点和 Kafka 端点的服务，这两个端点都应该采用请求对象的 JSON 表示形式（我们称之为 Foo）。我想确保 Foo 对象是有效的（通过 JSR-303 或其他）。所以 Foo 可能看起来像：

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

    // Getter and setter boilerplate
}

设置 REST 端点很简单：

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

如果我发帖，

{ "bar": 9 }

它会处理请求，但是如果我发帖：

{ "bar": 99 }

我得到一个错误的请求。到目前为止一切都很好！

Kafka 端点很容易创建（同时添加一个

StringJsonMessageConverter()

到我的

KafkaListenerContainerFactory

以便我得到 JSON-> 对象转换：

@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);
    }
}

但无论我尝试什么，我仍然可以传递无效的请求，并且它们处理无误。

我已经尝试了

@Valid

和

@Validated

。我试过添加

MethodValidationPostProcessor

豆。我试过向 KafkaListenerEndpointRegistrar 添加一个验证器（一个 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;
    }
}

我现在已经在这上面花了几天时间，我已经没有其他想法了。这甚至可能吗（无需将验证写入我的每个 kakfa 端点）？

Answer 1

抱歉耽搁了；本周我们在 SpringOne 平台上。

基础架构目前没有将

Validator

传递到有效载荷参数解析器中。请在 GitHub 上打开 issue.

Answer 2

Spring kafka 监听器默认不为非 Rest 控制器类扫描 @Valid。有关更多详细信息，请参阅此答案

https://stackoverflow.com/a/71859991/13898185

Answer 3

根据我的研究，Spring 使用 Jackson 将 json 转换为数组用于其内部处理，而 Spring Kafka 尚未支持它。 那么我们可以做的另一件事是使用 javax.validation 并手动定义它。 我就是这样做的，

private static final Validator validator;
static {
      validator= Validation.buildDefaultValidatorFactory().getValidator(); 
 }

@KafkaListener(topics = "fooTopic")
     public void kafkaEndpoint(Foo foo) {
        validator.validate(foo); //This will validate all the the json received


    

希望这有帮助