Forum Discussion

Anonymous's avatar
Anonymous
Not applicable
1 year ago

Event Hub Avro Data Fails to Deserialize in Fabric Eventstream

I am sending Avro-serialized messages from Kafka to Azure Event Hubs using SchemaRegistryApacheAvroSerializer. The schema is stored in Azure Schema Registry, and the producer sends events successfully.

public EventHubProducerService(
        VALUE("${EVENTHUB_CONNECTION_STRING}") String connectionString,
        VALUE("${eventhub.name}") String eventHubName,
        VALUE("${schema-registry.endpoint}") String schemaRegistryEndpoint,
        VALUE("${schema-registry.group}") String schemaRegistryGroup) {
    tokenCredential = new DefaultAzureCredentialBuilder().build();

    this.producerClient = new EventHubClientBuilder()
            .connectionString(connectionString, eventHubName)
            .buildProducerClient();

    SchemaRegistryAsyncClient schemaRegistryClient = new SchemaRegistryClientBuilder()
            .credential(tokenCredential)
            .fullyQualifiedNamespace(schemaRegistryEndpoint)
            .buildAsyncClient();

    this.schemaRegistryApacheAvroSerializer = new SchemaRegistryApacheAvroSerializerBuilder()
            .schemaRegistryClient(schemaRegistryClient)
            .schemaGroup(schemaRegistryGroup)
            .autoRegisterSchemas(true)
            .avroSpecificReader(true)
            .buildSerializer();
}

public void sendMessage(AgreementLifecycleDomainSourceType message) {
    EventData eventData = schemaRegistryApacheAvroSerializer.serialize(
            message, TypeReference.createInstance(EventData.class)
    );

    SendOptions sendOptions = new SendOptions().setPartitionId("1");
    producerClient.send(Collections.singletonList(eventData), sendOptions);
}

When I connect this Event Hub to Microsoft Fabric Eventstream and try to preview the data, I get this error:

Source 'EventHubInputAdapter' had occurrences of kind 'InputDeserializerError.InvalidData'. 
Invalid Avro Format.

Any insights would be really helpful!

eventstream

7 Replies

  • Anonymous's avatar
    Anonymous
    Not applicable
    1 year ago

    I just had a conversation with the Fabric Support team and they said Eventstream does not support reading from schema registry as of now.

  • nilendraFabric's avatar
    nilendraFabric
    Icon for Super User rankSuper User
    1 year ago

    Hello Anonymous 


    The error arises from Fabric’s inability to reconcile the producer’s schema with its own expectations.

     

    Fabric Eventstream expects the Avro schema used for deserialization to exactly match the schema used during serialization

    Ensure the schema used by your producer (`AgreementLifecycleDomainSourceType`) matches the schema Fabric expects. Use a schema validation tool or the Azure portal to confirm.

    Fabric uses the full format `namespace.servicebus.windows.net`, not just the namespace name.

     

     

     

    In

    • Anonymous's avatar
      Anonymous
      Not applicable
      1 year ago

      Thanks nilendraFabric for your response! I'd like to add few points.

      The schema used during serialization is the one which is being sent to Schema Registry. And I also tried deserializing the sent event data in my code and it deserializes just fine.

          public void sendMessage(AgreementLifecycleDomainSourceType message){
        EventData eventData = schemaRegistryApacheAvroSerializer.serialize(message, TypeReference.createInstance(EventData.class));
        System.out.println("Sent message to Event Hub: " + eventData.getBodyAsString());
        SendOptions sendOptions = new SendOptions().setPartitionId("1");
        List<EventData> serializedMessages = Collections.singletonList(eventData);
        producerClient.send(serializedMessages, sendOptions);
        for(EventData data : serializedMessages){
            AgreementLifecycleDomainSourceType out = schemaRegistryApacheAvroSerializer.deserialize(data, TypeReference.createInstance(AgreementLifecycleDomainSourceType.class));
            System.out.println("Deserialized data: "+out);
        }
    }


      And yes, I'm using the full format with '.servicebus.windows.net`


      Any idea where else it'd have gone wrong?

      • nilendraFabric's avatar
        nilendraFabric
        Icon for Super User rankSuper User
        1 year ago

        Try Bypass Schema Registry 
         Embed Schema: Serialize data with the Avro schema embedded (not using the registry) to ensure self-contained payloads.

         

        Something like this

         

        DatumWriter<AgreementLifecycleDomainSourceType> writer = new SpecificDatumWriter<>(schema);
        ByteArrayOutputStream outputStream = new ByteArrayOutputStream();
        Encoder encoder = EncoderFactory.get().binaryEncoder(outputStream, null);
        writer.write(message, encoder);
        encoder.flush();
        EventData eventData = new EventData(outputStream.toByteArray());