Configurare Hub eventi di Azure e endpoint del flusso di dati Kafka

Per configurare la comunicazione bidirezionale tra le operazioni IoT di Azure e i broker Apache Kafka, è possibile configurare un endpoint del flusso di dati. Questa configurazione consente di specificare l'endpoint, Transport Layer Security (TLS), l'autenticazione e altre impostazioni.

Prerequisiti

• Istanza delle operazioni di Azure IoT

Hub eventi di Azure

Hub eventi di Azure è compatibile con il protocollo Kafka e può essere usato con i flussi di dati con alcune limitazioni.

Creare uno spazio dei nomi e un hub eventi Hub eventi di Azure

Creare prima di tutto uno spazio dei nomi Hub eventi di Azure abilitato per Kafka

Creare quindi un hub eventi nello spazio dei nomi . Ogni singolo hub eventi corrisponde a un argomento Kafka. È possibile creare più hub eventi nello stesso spazio dei nomi per rappresentare più argomenti Kafka.

Assegnare l'autorizzazione all'identità gestita

Per configurare un endpoint del flusso di dati per Hub eventi di Azure, è consigliabile usare un'identità gestita assegnata dall'utente o assegnata dal sistema. Questo approccio è sicuro ed elimina la necessità di gestire manualmente le credenziali.

Dopo aver creato lo spazio dei nomi e l'hub eventi Hub eventi di Azure, è necessario assegnare un ruolo all'identità gestita di Operazioni IoT di Azure che concede l'autorizzazione per inviare o ricevere messaggi all'hub eventi.

Se si usa l'identità gestita assegnata dal sistema, in portale di Azure passare all'istanza di Operazioni IoT di Azure e selezionare Panoramica. Copiare il nome dell'estensione elencata dopo l'estensione Azure IoT Operations Arc. Ad esempio, azure-iot-operations-xxxx7. L'identità gestita assegnata dal sistema è disponibile usando lo stesso nome dell'estensione Azure IoT Operations Arc.

Passare quindi allo spazio dei nomi >di Hub eventi Controllo di accesso (IAM)>Aggiungi assegnazione di ruolo.

1. Nella scheda Ruolo selezionare un ruolo appropriato, ad esempio Azure Event Hubs Data Sender o Azure Event Hubs Data Receiver. In questo modo l'identità gestita ha le autorizzazioni necessarie per inviare o ricevere messaggi per tutti gli hub eventi nello spazio dei nomi. Per altre informazioni, vedere Autenticare un'applicazione con Microsoft Entra ID per accedere alle risorse di Hub eventi.
2. Nella scheda Membri :
   1. Se si usa l'identità gestita assegnata dal sistema, per Assegna accesso a, selezionare l'opzione Utente, gruppo o entità servizio, quindi selezionare + Seleziona membri e cercare il nome dell'estensione Azure IoT Operations Arc.
   2. Se si usa l'identità gestita assegnata dall'utente, per Assegna accesso a, selezionare l'opzione Identità gestita, quindi selezionare + Seleziona membri e cercare l'identità gestita assegnata dall'utente configurata per le connessioni cloud.

Creare un endpoint del flusso di dati per Hub eventi di Azure

Dopo aver configurato lo spazio dei nomi Hub eventi di Azure e l'hub eventi, è possibile creare un endpoint del flusso di dati per lo spazio dei nomi abilitato per Kafka Hub eventi di Azure.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param endpointName string = '<ENDPOINT_NAME>'
param hostName string = '<NAMESPACE>.servicebus.windows.net:9093'

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}
resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}
resource kafkaEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
  parent: aioInstance
  name: endpointName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    kafkaSettings: {
      host: hostName
      authentication: {
        // See available authentication methods section for method types
        // method: <METHOD_TYPE>
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

az deployment group create --resource-group <RESOURCE_GROUP> --template-file <FILE>.bicep

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT_NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <NAMESPACE>.servicebus.windows.net:9093
    authentication:
      # See available authentication methods section for method types
      # method: <METHOD_TYPE>
    tls:
      mode: Enabled

kubectl apply -f <FILE>.yaml

Usare stringa di connessione per l'autenticazione in Hub eventi

kafkaSettings: {
  authentication: {
    method: 'Sasl'
    saslSettings: {
      saslType: 'Plain'
      secretRef: '<SECRET_NAME>'
    }
  }
  tls: {
    mode: 'Enabled'
  }
}

kubectl create secret generic <SECRET_NAME> -n azure-iot-operations \
  --from-literal=username='$ConnectionString' \
  --from-literal=password='Endpoint=sb://<NAMESPACE>.servicebus.windows.net/;SharedAccessKeyName=<KEY-NAME>;SharedAccessKey=<KEY>'

kafkaSettings:
  authentication:
    method: Sasl
    saslSettings:
      saslType: Plain
      secretRef: <SECRET_NAME>
  tls:
    mode: Enabled

Limiti

Hub eventi di Azure non supporta tutti i tipi di compressione supportati da Kafka. Attualmente è supportata solo la compressione GZIP nei livelli premium e dedicati Hub eventi di Azure. L'uso di altri tipi di compressione potrebbe causare errori.

Broker Kafka personalizzati

Per configurare un endpoint del flusso di dati per broker Kafka non hub eventi, impostare l'host, TLS, l'autenticazione e altre impostazioni in base alle esigenze.

resource kafkaEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
 parent: aioInstance
  name: '<ENDPOINT NAME>'
  extendedLocation: {
    name: '<CUSTOM_LOCATION_NAME>'
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    host: '<KAFKA-HOST>:<PORT>'
    kafkaSettings: {
      authentication: {
        method: 'Sasl'
        saslSettings: {
            saslType: '<TYPE>'
            secretRef: '<SECRET_NAME>'
        }
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <KAFKA-HOST>:<PORT>
    authentication:
      method: Sasl
      saslSettings:
        saslType: <TYPE>
        secretRef: <SECRET_NAME>
    tls:
      mode: Enabled

Per personalizzare le impostazioni dell'endpoint, usare le sezioni seguenti per altre informazioni.

Metodi di autenticazione disponibili

Per gli endpoint del flusso di dati del broker Kafka sono disponibili i metodi di autenticazione seguenti.

Identità gestita assegnata dal sistema

Prima di configurare l'endpoint del flusso di dati, assegnare un ruolo all'identità gestita di Operazioni IoT di Azure che concede l'autorizzazione per connettersi al broker Kafka:

1. In portale di Azure passare all'istanza di Operazioni IoT di Azure e selezionare Panoramica.
2. Copiare il nome dell'estensione elencata dopo l'estensione Azure IoT Operations Arc. Ad esempio, azure-iot-operations-xxxx7.
3. Passare alla risorsa cloud necessaria per concedere le autorizzazioni. Ad esempio, passare allo spazio dei nomi >di Hub eventi Controllo di accesso (IAM)>Aggiungi assegnazione di ruolo.
4. Nella scheda Ruolo selezionare un ruolo appropriato.
5. Nella scheda Membri , per Assegna accesso a, selezionare l'opzione Utente, gruppo o entità servizio, quindi selezionare + Seleziona membri e cercare l'identità gestita di Operazioni IoT di Azure. Ad esempio, azure-iot-operations-xxxx7.

Configurare quindi l'endpoint del flusso di dati con le impostazioni di identità gestite assegnate dal sistema.

kafkaSettings: {
  authentication: {
    method: 'SystemAssignedManagedIdentity'
    systemAssignedManagedIdentitySettings: {}
  }
}

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      {}

Questa configurazione crea un'identità gestita con il gruppo di destinatari predefinito, che corrisponde al valore host dello spazio dei nomi di Hub eventi sotto forma di https://<NAMESPACE>.servicebus.windows.net. Tuttavia, se è necessario eseguire l'override del gruppo di destinatari predefinito, è possibile impostare il audience campo sul valore desiderato.

kafkaSettings: {
  authentication: {
    method: 'SystemAssignedManagedIdentity'
    systemAssignedManagedIdentitySettings: {
        audience: '<YOUR_AUDIENCE_OVERRIDE_VALUE>'
    }
  }
}

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      audience: <YOUR_AUDIENCE_OVERRIDE_VALUE>

Identità gestita assegnata dall'utente

Per usare l'identità gestita assegnata dall'utente per l'autenticazione, è prima necessario distribuire le operazioni IoT di Azure con impostazioni sicure abilitate. È quindi necessario configurare un'identità gestita assegnata dall'utente per le connessioni cloud. Per altre informazioni, vedere Abilitare le impostazioni sicure nella distribuzione di Operazioni IoT di Azure.

Prima di configurare l'endpoint del flusso di dati, assegnare un ruolo all'identità gestita assegnata dall'utente che concede l'autorizzazione per connettersi al broker Kafka:

1. In portale di Azure passare alla risorsa cloud necessaria per concedere le autorizzazioni. Ad esempio, passare allo spazio dei nomi >Griglia di eventi Controllo di accesso (IAM)>Aggiungi assegnazione di ruolo.
2. Nella scheda Ruolo selezionare un ruolo appropriato.
3. Nella scheda Membri selezionare l'opzione Identità gestita per Assegnare l'accesso, quindi selezionare + Seleziona membri e cercare l'identità gestita assegnata dall'utente.

Configurare quindi l'endpoint del flusso di dati con le impostazioni di identità gestite assegnate dall'utente.

kafkaSettings: {
  authentication: {
    method: 'UserAssignedManagedIdentity'
    UserAssignedManagedIdentitySettings: {
      clientId: '<CLIENT_ID>'
      tenantId: '<TENANT_ID>'
      // Optional, defaults to https://<NAMESPACE>.servicebus.windows.net/.default
      // Matching the Event Hub namespace you configured as host
      // scope: 'https://<SCOPE_URL>'
    }
  }
  ...
}

kafkaSettings:
  authentication:
    method: UserAssignedManagedIdentity
    userAssignedManagedIdentitySettings:
      clientId: <CLIENT_ID>
      tenantId: <TENANT_ID>
      # Optional, defaults to https://<NAMESPACE>.servicebus.windows.net/.default
      # Matching the Event Hub namespace you configured as host
      # scope: https://<SCOPE_URL>

In questo caso, l'ambito è il gruppo di destinatari dell'identità gestita. Il valore predefinito è uguale al valore host dello spazio dei nomi di Hub eventi sotto forma di https://<NAMESPACE>.servicebus.windows.net. Tuttavia, se è necessario eseguire l'override del gruppo di destinatari predefinito, è possibile impostare il campo ambito sul valore desiderato usando Bicep o Kubernetes.

SASL

Per usare SASL per l'autenticazione, specificare il metodo di autenticazione SASL e configurare il tipo SASL e un riferimento segreto con il nome del segreto che contiene il token SASL.

kafkaSettings: {
  authentication: {
    method: 'Sasl' // Or ScramSha256, ScramSha512
    saslSettings: {
      saslType: 'Plain' // Or ScramSha256, ScramSha512
      secretRef: '<SECRET_NAME>'
    }
  }
}

kubectl create secret generic sasl-secret -n azure-iot-operations \
  --from-literal=token='<YOUR_SASL_TOKEN>'

kafkaSettings:
  authentication:
    method: Sasl
    saslSettings:
      saslType: Plain # Or ScramSha256, ScramSha512
      secretRef: <SECRET_NAME>

I tipi SASL supportati sono:

• Plain
• ScramSha256
• ScramSha512

Il segreto deve trovarsi nello stesso spazio dei nomi dell'endpoint del flusso di dati Kafka. Il segreto deve avere il token SASL come coppia chiave-valore.

Anonimo

Per usare l'autenticazione anonima, aggiornare la sezione di autenticazione delle impostazioni Kafka per usare il metodo Anonimo.

kafkaSettings: {
  authentication: {
    method: 'Anonymous'
  }
}

kafkaSettings:
  authentication:
    method: Anonymous
    anonymousSettings:
      {}

Impostazioni avanzate

È possibile impostare impostazioni avanzate per l'endpoint del flusso di dati Kafka, ad esempio TLS, certificato CA attendibile, impostazioni di messaggistica Kafka, invio in batch e CloudEvents. È possibile impostare queste impostazioni nella scheda Del portale avanzato dell'endpoint del flusso di dati o nella risorsa endpoint del flusso di dati.

// See sections below for more details
kafkaSettings: {
  consumerGroupId: '<ID>'
  compression: 'Gzip'
  copyMqttProperties: 'Disabled'
  kafkaAcks: 'All'
  partitionStrategy: 'Default'
  tls: {
    mode: 'Enabled'
    trustedCaCertificateConfigMapRef: '<YOUR_CA_CERTIFICATE>'
  }
  batching: {
    mode: 'Enabled'
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  }
}

# See sections below for more details
kafkaSettings:
  consumerGroupId: <ID>
  compression: Gzip
  copyMqttProperties: Disabled
  kafkaAcks: All
  partitionStrategy: Default
  tls:
    mode: Enabled
    trustedCaCertificateConfigMapRef: <YOUR_CA_CERTIFICATE>
  batching:
    mode: Enabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024

Impostazioni di TLS

Modalità TLS

Per abilitare o disabilitare TLS per l'endpoint Kafka, aggiornare l'impostazione mode nelle impostazioni TLS.

kafkaSettings: {
  tls: {
    mode: 'Enabled' // Or Disabled
  }
}

kafkaSettings:
  tls:
    mode: Enabled # Or Disabled

La modalità TLS può essere impostata su Enabled o Disabled. Se la modalità è impostata su Enabled, il flusso di dati usa una connessione sicura al broker Kafka. Se la modalità è impostata su Disabled, il flusso di dati usa una connessione non sicura al broker Kafka.

Certificato CA attendibile

Configurare il certificato CA attendibile per l'endpoint Kafka per stabilire una connessione sicura al broker Kafka. Questa impostazione è importante se il broker Kafka usa un certificato autofirmato o un certificato firmato da una CA personalizzata non considerata attendibile per impostazione predefinita.

kafkaSettings: {
  tls: {
    trustedCaCertificateConfigMapRef: '<YOUR_CA_CERTIFICATE>'
  }
}

kafkaSettings:
  tls:
    trustedCaCertificateConfigMapRef: <YOUR_CA_CERTIFICATE>

Questo ConfigMap deve contenere il certificato della CA in formato PEM. ConfigMap deve trovarsi nello stesso spazio dei nomi della risorsa del flusso di dati Kafka. Ad esempio:

kubectl create configmap client-ca-configmap --from-file root_ca.crt -n azure-iot-operations

ID gruppo di consumer

L'ID del gruppo di consumer viene usato per identificare il gruppo di consumer usato dal flusso di dati per leggere i messaggi dall'argomento Kafka. L'ID del gruppo di consumer deve essere univoco all'interno del broker Kafka.

kafkaSettings: {
  consumerGroupId: '<ID>'
}

spec:
  kafkaSettings:
    consumerGroupId: <ID>

Questa impostazione ha effetto solo se l'endpoint viene usato come origine, ovvero il flusso di dati è un consumer.

Compressione

Il campo di compressione abilita la compressione per i messaggi inviati agli argomenti Kafka. La compressione consente di ridurre la larghezza di banda di rete e lo spazio di archiviazione necessario per il trasferimento dei dati. Tuttavia, la compressione aggiunge anche un sovraccarico e una latenza al processo. I tipi di compressione supportati sono elencati nella tabella seguente.

Per configurare la compressione:

kafkaSettings: {
  compression: 'Gzip' // Or Snappy, Lz4
}

kafkaSettings:
  compression: Gzip # Or Snappy, Lz4

Questa impostazione ha effetto solo se l'endpoint viene usato come destinazione in cui il flusso di dati è un producer.

Batch

Oltre alla compressione, è anche possibile configurare l'invio in batch per i messaggi prima di inviarli agli argomenti Kafka. L'invio in batch consente di raggruppare più messaggi e comprimerli come singola unità, migliorando così l'efficienza di compressione e riducendo il sovraccarico di rete.

Ad esempio, se si impostano i messaggi latencym su 1000, maxMessages su 100 e maxBytes su 1024, i messaggi vengono inviati quando sono presenti 100 messaggi nel buffer o quando sono presenti 1.024 byte nel buffer o quando sono trascorsi 1.000 millisecondi dall'ultimo invio, a ogni occorrenza.

Per configurare l'invio in batch:

kafkaSettings: {
  batching: {
    mode: 'Enabled' // Or Disabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  }
}

kafkaSettings:
  batching:
    mode: Enabled # Or Disabled
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024

Questa impostazione ha effetto solo se l'endpoint viene usato come destinazione in cui il flusso di dati è un producer.

Strategia di gestione delle partizioni

La strategia di gestione delle partizioni controlla il modo in cui i messaggi vengono assegnati alle partizioni Kafka durante l'invio agli argomenti Kafka. Le partizioni Kafka sono segmenti logici di un argomento Kafka che consentono l'elaborazione parallela e la tolleranza di errore. Ogni messaggio in un argomento Kafka ha una partizione e un offset, che vengono usati per identificare e ordinare i messaggi.

Questa impostazione ha effetto solo se l'endpoint viene usato come destinazione in cui il flusso di dati è un producer.

Per impostazione predefinita, un flusso di dati assegna messaggi a partizioni casuali, usando un algoritmo round robin. Tuttavia, è possibile usare strategie diverse per assegnare messaggi alle partizioni in base ad alcuni criteri, ad esempio il nome dell'argomento MQTT o una proprietà del messaggio MQTT. Ciò consente di ottenere un migliore bilanciamento del carico, località dei dati o ordinamento dei messaggi.

Ad esempio, se si imposta la strategia di gestione delle partizioni su Property e la proprietà della chiave di partizione su device-id, i messaggi con la stessa proprietà vengono inviati alla stessa device-id partizione.

Per configurare la strategia di gestione delle partizioni:

kafkaSettings: {
  partitionStrategy: 'Default' // Or Static, Topic, Property
  partitionKeyProperty: '<PROPERTY_NAME>'
}

kafkaSettings:
  partitionStrategy: Default # Or Static, Topic, Property
  partitionKeyProperty: <PROPERTY_NAME>

Riconoscimenti Kafka

I riconoscimenti Kafka (ack) vengono usati per controllare la durabilità e la coerenza dei messaggi inviati agli argomenti Kafka. Quando un producer invia un messaggio a un argomento Kafka, può richiedere diversi livelli di riconoscimenti dal broker Kafka per assicurarsi che il messaggio venga scritto correttamente nell'argomento e replicato nel cluster Kafka.

Questa impostazione ha effetto solo se l'endpoint viene usato come destinazione, ovvero il flusso di dati è un producer.

Ad esempio, se si imposta il riconoscimento Kafka su All, il flusso di dati attende che il messaggio venga scritto nella partizione leader e in tutte le partizioni di follower prima di inviare il messaggio successivo.

Per configurare i riconoscimenti Kafka:

kafkaSettings: {
  kafkaAcks: 'All' // Or None, One, Zero
}

kafkaSettings:
  kafkaAcks: All # Or None, One, Zero

Questa impostazione ha effetto solo se l'endpoint viene usato come destinazione in cui il flusso di dati è un producer.

Copiare le proprietà MQTT

Per impostazione predefinita, l'impostazione delle proprietà MQTT di copia è abilitata. Queste proprietà utente includono valori come subject, che archivia il nome dell'asset che invia il messaggio.

kafkaSettings: {
  copyMqttProperties: 'Enabled' // Or Disabled
}

kafkaSettings:
  copyMqttProperties: Enabled # Or Disabled

Le sezioni seguenti descrivono come le proprietà MQTT vengono convertite in intestazioni utente Kafka e viceversa quando l'impostazione è abilitata.

L'endpoint Kafka è una destinazione

Quando un endpoint Kafka è una destinazione del flusso di dati, tutte le proprietà definite dalla specifica MQTT v5 vengono convertite nelle intestazioni utente Kafka. Ad esempio, un messaggio MQTT v5 con "Tipo di contenuto" inoltrato a Kafka si traduce nell'intestazione "Content Type":{specifiedValue}utente Kafka . Le regole simili si applicano ad altre proprietà MQTT predefinite, definite nella tabella seguente.

Le coppie chiave chiave della proprietà utente MQTT v5 vengono convertite direttamente in intestazioni utente Kafka. Se un'intestazione utente in un messaggio ha lo stesso nome di una proprietà MQTT predefinita (ad esempio, un'intestazione utente denominata "Dati di correlazione"), se inoltrare il valore della proprietà della specifica MQTT v5 o la proprietà utente non è definita.

I flussi di dati non ricevono mai queste proprietà da un broker MQTT. Di conseguenza, un flusso di dati non li inoltra mai:

• Alias dell'argomento
• Identificatori di sottoscrizione

Proprietà Message Expiry Interval

L'intervallo di scadenza del messaggio specifica per quanto tempo un messaggio può rimanere in un broker MQTT prima di essere rimosso.

Quando un flusso di dati riceve un messaggio MQTT con l'intervallo di scadenza del messaggio specificato, è possibile:

• Registra l'ora di ricezione del messaggio.
• Prima che il messaggio venga generato nella destinazione, l'ora viene sottratta dal messaggio dall'intervallo di scadenza originale.
• Se il messaggio non è scaduto (l'operazione precedente è > 0), il messaggio viene generato nella destinazione e contiene l'ora di scadenza del messaggio aggiornata.
• Se il messaggio è scaduto (l'operazione precedente è <= 0), il messaggio non viene generato dalla destinazione.

Esempi:

• Un flusso di dati riceve un messaggio MQTT con intervallo di scadenza messaggio = 3600 secondi. La destinazione corrispondente è temporaneamente disconnessa, ma è in grado di riconnettersi. 1.000 secondi passano prima che questo messaggio MQTT venga inviato alla destinazione. In questo caso, il messaggio di destinazione ha l'intervallo di scadenza del messaggio impostato su 2600 (3600 - 1000) secondi.
• Il flusso di dati riceve un messaggio MQTT con intervallo di scadenza del messaggio = 3600 secondi. La destinazione corrispondente è temporaneamente disconnessa, ma è in grado di riconnettersi. In questo caso, tuttavia, la riconnessione richiede 4.000 secondi. Il messaggio è scaduto e il flusso di dati non inoltra questo messaggio alla destinazione.

L'endpoint Kafka è un'origine del flusso di dati

Quando un endpoint Kafka è un'origine del flusso di dati, le intestazioni utente Kafka vengono convertite in proprietà MQTT v5. Nella tabella seguente viene descritto il modo in cui le intestazioni utente Kafka vengono convertite in proprietà MQTT v5.

Coppie chiave/valore dell'intestazione utente Kafka, purché siano tutte codificate in UTF-8, vengono convertite direttamente in proprietà chiave/valore utente MQTT.

UTF-8/ Mancata corrispondenza binaria

MQTT v5 può supportare solo le proprietà basate su UTF-8. Se il flusso di dati riceve un messaggio Kafka che contiene una o più intestazioni non UTF-8, il flusso di dati:

• Rimuovere la proprietà o le proprietà che causano un'offesa.
• Inoltrare il resto del messaggio, seguendo le regole precedenti.

Le applicazioni che richiedono il trasferimento binario nelle intestazioni di origine Kafka => le proprietà della destinazione MQTT devono prima codificarle in UTF-8, ad esempio tramite Base64.

>=64 KB proprietà non corrispondenti

Le proprietà MQTT v5 devono essere inferiori a 64 KB. Se il flusso di dati riceve un messaggio Kafka che contiene una o più intestazioni = >64 KB, il flusso di dati:

• Rimuovere la proprietà o le proprietà che causano un'offesa.
• Inoltrare il resto del messaggio, seguendo le regole precedenti.

Conversione delle proprietà quando si usano Hub eventi e producer che usano AMQP

Se si dispone di un client che inoltra messaggi a un endpoint dell'origine del flusso di dati Kafka, eseguendo una delle azioni seguenti:

• Invio di messaggi a Hub eventi tramite librerie client come Azure.Messaging.EventHubs
• Uso diretto di AMQP

Esistono sfumature di traduzione delle proprietà da tenere presenti.

È consigliabile eseguire una delle operazioni seguenti:

• Evitare l'invio di proprietà
• Se è necessario inviare proprietà, inviare valori codificati come UTF-8.

Quando Hub eventi converte le proprietà da AMQP a Kafka, include i tipi codificati AMQP sottostanti nel messaggio. Per altre informazioni sul comportamento, vedere Scambio di eventi tra consumer e producer che usano protocolli diversi.

Nell'esempio di codice seguente quando l'endpoint del flusso di dati riceve il valore "foo":"bar", riceve la proprietà come <0xA1 0x03 "bar">.

using global::Azure.Messaging.EventHubs;
using global::Azure.Messaging.EventHubs.Producer;

var propertyEventBody = new BinaryData("payload");

var propertyEventData = new EventData(propertyEventBody)
{
  Properties =
  {
    {"foo", "bar"},
  }
};

var propertyEventAdded = eventBatch.TryAdd(propertyEventData);
await producerClient.SendAsync(eventBatch);

L'endpoint del flusso di dati non può inoltrare la proprietà <0xA1 0x03 "bar"> payload a un messaggio MQTT perché i dati non sono UTF-8. Tuttavia, se si specifica una stringa UTF-8, l'endpoint del flusso di dati converte la stringa prima dell'invio a MQTT. Se si usa una stringa UTF-8, il messaggio MQTT avrà "foo":"bar" come proprietà utente.

Vengono convertite solo le intestazioni UTF-8. Ad esempio, dato lo scenario seguente in cui la proprietà è impostata come float:

Properties = 
{
  {"float-value", 11.9 },
}

L'endpoint del flusso di dati elimina i pacchetti che contengono il "float-value" campo .

Non tutte le proprietà dei dati dell'evento, inclusa propertyEventData.correlationId, vengono inoltrate. Per altre informazioni, vedere Proprietà utente evento,

CloudEvents

CloudEvents è un modo per descrivere i dati degli eventi in modo comune. Le impostazioni cloudEvents vengono usate per inviare o ricevere messaggi nel formato CloudEvents. È possibile usare CloudEvents per architetture basate su eventi in cui servizi diversi devono comunicare tra loro nello stesso provider di servizi cloud o diversi.

Le CloudEventAttributes opzioni sono Propagate oCreateOrRemap.

kafkaSettings: {
  cloudEventAttributes: 'Propagate' // Or CreateOrRemap
}

kafkaSettings:
  cloudEventAttributes: Propagate # Or CreateOrRemap

Le sezioni seguenti descrivono in che modo le proprietà CloudEvent vengono propagate o create e ricreate e mappate.

Impostazione di propagazione

Le proprietà CloudEvent vengono passate per i messaggi che contengono le proprietà necessarie. Se il messaggio non contiene le proprietà necessarie, il messaggio viene passato così come è. Se sono presenti le proprietà necessarie, viene aggiunto un ce_ prefisso al nome della proprietà CloudEvent.

Impostazione CreateOrRemap

Le proprietà CloudEvent vengono passate per i messaggi che contengono le proprietà necessarie. Se il messaggio non contiene le proprietà necessarie, le proprietà vengono generate.

Passaggi successivi

Per altre informazioni sui flussi di dati, vedere Creare un flusso di dati.