Printemps Cloud Dataflow (lapin | processeur | hdfs) binaire de sortie

Question

Actuellement, j'ai un cas d'utilisation où j'ai besoin de prendre des messages de RabbitMQ Message Bus, ajouter la taille du message (en octets) et le message de sortie en utilisant HDFS Sink. Pour commencer, j'ai créé mon propre processeur qui ajoute la taille au message. La raison pour laquelle je fais cela est parce que l'encodage doit être celui d'un tampon de protocole de Google.

Mon application se présente comme suit:

stream create --name rabbit-to-hdfs --definition "rabbit | delim-protobuf | hdfs " 

Lorsque HDFS Sink émet le message que je vois [B @ 12768762. Je Google'd autour et ai vu des recommandations pour ajouter ce qui suit:

spring.cloud.stream.bindings.input.consumer.headerMode=raw 

Cependant, cela ne voit pas me aider du tout! Cela dit, si je change l'application pour aller à un fichier en utilisant les éléments suivants:

[input | processor ] | file --binary=true 

Ensuite, tout fonctionne très bien. Cependant, j'aime les fonctionnalités de retournement offertes par le HDFS Sink.

Des idées?

Answer 1

Le fichier fonctionne car il ne fait que déverser les octets reçus, mais en regardant le récepteur HDFS, il semble qu'il doive consommer un objet java.io.Serializable comme entrée. Mais dans votre cas, vous envoyez un tableau d'octets à partir d'un objet protobuf (je suppose que c'est ce qui se passe)

Answer 2

Les types ne sont pas compatibles c'est le problème. En définissant contentType dans SCS, vous demandez simplement au framework d'utiliser la sérialisation java pour appeler writeObject. Mais puisque vous utilisez protobuf qui est déjà un framework de sérialisation, ça ne marcherait pas. Le problème ici est que le récepteur semble vraiment (je ne suis pas familier avec le code de l'évier) pour s'attendre à un Serialisable, mais vous n'en fournissez pas. Qu'est-ce que vous pouvez faire est de modifier l'application de l'évier ou fournir un convertisseur personnalisé qui sait comment convertir de protobuf en Serializable, ne sais même pas si cela a du sens pour être honnête.