KafkaAvroSerializer 的额外字节

Question

我的设置如下：我从 ftp 服务器检索 xml 文件，将它们解组到 POJO，将其映射到 Avro 生成的类，然后将其转发到 Alpakkas's Producer Sink，如下所示：

Ftp.ls("/", ftpSettings)
  .filter(FtpFile::isFile)
  .mapAsyncUnordered(10,
    ftpFile -> {
      CompletionStage<ByteString> fetchFile =
        Ftp.fromPath(ftpFile.path(), ftpSettings).runWith(Sink.reduce((a, b) -> a), materializer);
      return fetchFile;
    })
  .map(b -> b.decodeString(Charsets.ISO_8859_1))
  .map(StringReader::new)
  .map(AlpakkaProducerDemo::unmarshalFile)
  .map(AlpakkaProducerDemo::convertToAvroSerializable)
  .map(a -> new ProducerRecord<>(kafkaTopic, a.id().toString(), a))
  .map(record -> ProducerMessage.single(record))
  .runWith(Producer.committableSink(producerSettings, kafkaProducer), materializer);

问题是序列化显然不能正常工作。例如。我也希望将密钥进行 avro 序列化，尽管它只是一个字符串（要求，不要问）。配置如下：

Map<String, Object> kafkaAvroSerDeConfig = new HashMap<>();
kafkaAvroSerDeConfig.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, schemaRegistryUrl);
final KafkaAvroSerializer keyAvroSerializer = new KafkaAvroSerializer();
keyAvroSerializer.configure(kafkaAvroSerDeConfig, true);
final Serializer<Object> keySerializer = keyAvroSerializer;
final Config config = system.settings().config().getConfig("akka.kafka.producer");
final ProducerSettings producerSettings = ProducerSettings.create(config, keySerializer, valueSerializer)
  .withBootstrapServers(kafkaServer);

在 Kafka 中，这会产生一个内容正确的键，但字符串开头有一些（明显的）额外字节：\u0000\u0000\u0000\u0000\u0001N。可以想象，这会对价值造成严重破坏。我怀疑 Avro 序列化不适合 Alpakka 使用的信封 API，因此可能需要事先序列化为 byte[] 并使用常见的 ByteSerializer。但是，那时使用SchemaRegistry 没有任何意义。

Answer 1

前五个字节与序列化格式版本（字节 0）和架构注册表中的 Avro 架构版本（字节 1-4）有关：https://docs.confluent.io/current/schema-registry/serializer-formatter.html#wire-format。

另一种选择可能是使用 Kafka Connect，以及 FTP 源和 XML 转换。