无法使用 Nifi 和 Schema Registry 使用 Kafka Avro 记录

Question

我正在尝试使用 Nifi 使用来自 Kafka 的 Avro 记录。我有 3 个来自 Amazon Lambda 的主题和 2 个 Spark Streaming 作业，所有这些作业都使用 HortonWorks Schema Registry 来获取 Avro 模式。

我尝试使用 ConsumeKafkaRecord_0_10 和 ConsumeKafkaRecord_2_0 并得到相同的错误：

我尝试使用 AvroReader 在内部使用纯文本模式，以确保正在使用的模式，并得到了同样的错误。 当我使用带有 Horton Schema Registry 参数的 AvroReader 时，我收到此错误：

这很有意义，因为它将记录的第一个字节作为模式的版本参数，而第一个字节是 3。但这并不能解释为什么我在将模式放在普通模式时会得到 ArrayIndexOutOfBound文本。

最后我可以使用 Spark Streaming 和 Schema Registry 很好地使用这些主题。在使用 Kafka 时，没有人在 NiFi 和 AvroReader 之间遇到过这样的问题吗？

堆栈：Horton Works HDP 3.4.1 // Nifi 1.9.0 // Spark 2.3 // Schema Registry 0.7

Answer 1

该问题与 Nifi 如何解释 Avro 消息的第一个字节有关。这些字节包含以下信息：

• 协议 ID - 1 字节
• 架构元数据 ID - 8 个字节
• 架构版本 - 4 个字节

通过HortonWork Schema Registry的代码，我们可以发现不同的Protocol ID可以用来序列化你的AvroSerDe消息。

public static final byte CONFLUENT_VERSION_PROTOCOL = 0x0;
public static final byte METADATA_ID_VERSION_PROTOCOL = 0x1;
public static final byte VERSION_ID_AS_LONG_PROTOCOL = 0x2;
public static final byte VERSION_ID_AS_INT_PROTOCOL = 0x3;
public static final byte CURRENT_PROTOCOL = VERSION_ID_AS_INT_PROTOCOL;

Source

默认使用的是 VERSION_ID_AS_INT_PROTOCOL，这意味着 Avro 消息的第一个字节将是 03。

在查看 Nifi 代码时，我们看到它实际上只使用 METADATA_ID_VERSION_PROTOCOL，期待 01 而没有考虑其他任何内容。

在创建 SchemaRegistryConfig 时，您必须强制 Spark 使用 METADATA_ID_VERSION_PROTOCOL。

  val config = Map[String, Object](
    "schema.registry.url" -> ConfigManager.config.getProperty("schemaregistry.default.url"),
    AbstractAvroSnapshotSerializer.SERDES_PROTOCOL_VERSION -> SerDesProtocolHandlerRegistry.METADATA_ID_VERSION_PROTOCOL.asInstanceOf[Object]
  )
  implicit val srConfig:SchemaRegistryConfig = SchemaRegistryConfig(config)