java - Azure Event Hubs Avro Data Fails to Deserialize in Fabric Eventstream - Stack Overflow

admin管理员组

文章数量:1278910

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 helpful!

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 helpful!

• java
• azure
• azure-eventhub
• microsoft-fabric
• azure-schemaregistry

Share Improve this question Follow asked Feb 24 at 17:24 hamamhamam 5577 bronze badges 1

• Use this sample code to Serialize ``` PlayingCard playingCard = new PlayingCard(); playingCard.setPlayingCardSuit(PlayingCardSuit.SPADES); playingCard.setIsFaceCard(false); playingCard.setCardValue(5); MessageContent message = serializer.serialize(playingCard, TypeReference.createInstance(MessageContent.class));``` – Sampath Commented Feb 28 at 4:23

Add a comment  | 

1 Answer 1

Sorted by: Reset to default 0

The Invalid Avro Format error that occurs when trying to deserialize Avro messages in Microsoft Fabric Eventstream from Azure Event Hubs can be resolved by using the following code:

PlayingCard playingCard = new PlayingCard();
playingCard.setPlayingCardSuit(PlayingCardSuit.SPADES);
playingCard.setIsFaceCard(false);
playingCard.setCardValue(5);

MessageContent message = serializer.serialize(playingCard,
    TypeReference.createInstance(MessageContent.class));

In this case, change .avroSpecificReader(true) to .avroSpecificReader(false).

Below is a sample code snippet for the Azure Schema Registry Apache Avro Serializer client library for Java with Azure Event Hubs. Refer to this GitHub repository for sample code on serialization and deserialization:

private final EventHubProducerClient producerClient;
private final SchemaRegistryApacheAvroSerializer schemaRegistryApacheAvroSerializer;
private final TokenCredential tokenCredential;

public EventHubProducerService(
        String connectionString,
        String eventHubName,
        String schemaRegistryEndpoint,
        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(false)  
            .buildSerializer();
}

public void sendMessage(GenericRecord message) {
    try {
        EventData eventData = schemaRegistryApacheAvroSerializer.serialize(
                message, TypeReference.createInstance(EventData.class)
        );
        System.out.println("Serialized Message: " + new String(eventData.getBody()));

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

        System.out.println("Message successfully sent to Event Hub.");
    } catch (Exception e) {
        System.err.println("Error sending Avro message: " + e.getMessage());
        e.printStackTrace();
    }
}

本文标签： javaAzure Event Hubs Avro Data Fails to Deserialize in Fabric EventstreamStack Overflow