1

在 docker 中作为独立运行 Pulsar 时，在特定情况下反序列化消息时，我们会遇到这个奇怪的问题。我们使用的是 2.7.1 版本。

我们有一个创建主题和函数的脚本，然后为 JSON 类型的麻烦主题创建模式。整个架构是正确的，但类型不正确。这都是在发送任何消息之前。我们还启用了set-is-allow-auto-update-schema.

我们称之为它trouble-topic，它由 2 个来源填充：ValidationFunction和一个 Spring Boot 微服务。

ValidationFunction验证消息，如果消息有效，它将映射消息发送到 Spring Boot 微服务使用的主题，然后对其执行一些逻辑并将其发送到trouble-topic，但如果验证失败，则直接将消息发送到trouble-topic。

当sendAsync从 Spring Boot 微服务与以下生产者一起使用时，模式得到更新，将 AVRO 作为一种类型，然后TroubleFunction读取trouble-topic工作正常：

pulsarClient
    .newProducer(AvroSchema.of(OurClass.class))
    .topic(troubleTopicName))
    .create()

但是如果在此之前有些消息没有通过验证，并且在使用上面的 Producer 之前直接发送消息trouble-topic，我们就会得到一个解析异常。我们通过以下方式从函数发送消息：

context.newOutputMessage(troubleTopicName, AvroSchema.of(OurClass.class))
    .value(value)
    .sendAsync();

由于某种原因，这不会更新架构类型，并且架构类型仍然是 JSON。我使用 pulsar admin CLI 验证了每个步骤的模式类型。当这种情况发生在微服务生产者第一次更新模式类型之前，TroubleFunction读取trouble-topic失败并出现以下错误：

11:43:49.322 [tenant/namespace/TroubleFunction-0] ERROR org.apache.pulsar.functions.instance.JavaInstanceRunnable - [tenant/namespace/TroubleFunction:0] Uncaught exception in Java Instance
org.apache.pulsar.client.api.SchemaSerializationException: com.fasterxml.jackson.core.JsonParseException: Illegal character ((CTRL-CHAR, code 2)): only regular white space (\r, \n, \t) is allowed between tokens
 at [Source: (byte[])avro-serialized-msg-i-have-to-hide Parsing exception: cvc-complex-type.2.4.a: Invalid content was found starting with element 'ElementName'. One of '{"foo:bar":ElementName}' is expected."; line: 1, column: 2]

所以我的问题是这两者有什么区别，为什么从函数发送消息不能正确更新模式类型？不是在下面使用同一个 Producer 吗？还有一种方法可以解决这个问题，以便在初始化时设置模式类型，或者至少在从函数发送消息时更新模式类型？

4

1 回答 1

2

首先，信用到期。我想有一天这会被很好地记录下来，但现在还没有。我很幸运拥有 Apache Pulsar in Action 的 EAP 版本，其中这个示例存储库用于展示 Pulsar 的一些功能：https ://github.com/david-streamlio/GottaEat

我强烈推荐这本书，并为每个使用 Pulsar 工作的人浏览这些示例，在 pulsar slack 社区上有人提到，就在昨天它从 MEAP 毕业，它应该很快就会有印刷版，所以请检查一下。也可以考虑加入 Pulsar slack。

回答：

这是让我了解它应该如何工作的代码：

Map<String, ConsumerConfig> inputSpecs = new HashMap<String, ConsumerConfig> ();
inputSpecs.put("persistent://orders/inbound/food-orders", 
    ConsumerConfig.builder().schemaType("avro").build());
FunctionConfig functionConfig = 
    FunctionConfig.builder()
        ...
        .inputSpecs(inputSpecs)
        ...
        .build();

使用 LocalRunner 时可以使用 Java 代码来设置功能，但使用 pulsar admin cli（我们使用）和 rest api 可以实现相同的配置。您也可以使用函数配置文件，并在配置 yaml 中通过以下方式指定它：

inputSpecs:
 $topicName:
  schemaType: AVRO

$topicName总是采用以下格式：persistent://tenant/namespace/topic

在我的情况下，一旦为 指定输入规范，TroubleFunction将使用正确的模式类型有效地创建模式，并且反序列化也可以正常工作。

于 2021-11-02T14:20:03.860 回答