Inviare e ricevere messaggi tra l'anteprima mq di Azure IoT e Hub eventi di Azure o Kafka

Importante

Anteprima delle operazioni di Azure IoT: abilitata da Azure Arc è attualmente disponibile in ANTEPRIMA. Non è consigliabile usare questo software di anteprima negli ambienti di produzione.

Vedere le condizioni per l'utilizzo supplementari per le anteprime di Microsoft Azure per termini legali aggiuntivi che si applicano a funzionalità di Azure in versione beta, in anteprima o in altro modo non ancora disponibili a livello generale.

Il connettore Kafka esegue il push dei messaggi dal broker MQ Preview di Azure IoT MQ a un endpoint Kafka e esegue il pull dei messaggi in altro modo. Poiché Hub eventi di Azure supporta l'API Kafka, il connettore funziona in modo predefinito con Hub eventi.

Configurare il connettore di Hub eventi tramite l'endpoint Kafka

Per impostazione predefinita, il connettore non è installato con Azure IoT MQ. Deve essere abilitata in modo esplicito con il mapping degli argomenti e le credenziali di autenticazione specificate. Seguire questa procedura per abilitare la comunicazione bidirezionale tra IoT MQ e Hub eventi di Azure tramite l'endpoint Kafka.

1. Creare uno spazio dei nomi di Hub eventi.

2. Creare un hub eventi per ogni argomento Kafka.

Concedere al connettore l'accesso allo spazio dei nomi di Hub eventi

La concessione dell'accesso all'estensione MQ Arc di IoT a uno spazio dei nomi di Hub eventi è il modo più pratico per stabilire una connessione sicura dal connettore Kakfa di IoT MQ a Hub eventi.

Salvare il modello Bicep seguente in un file e applicarlo con l'interfaccia della riga di comando di Azure dopo aver impostato i parametri validi per l'ambiente:

Nota

Il modello Bicep presuppone che il cluster connnected Arc e lo spazio dei nomi di Hub eventi si trovino nello stesso gruppo di risorse, modificare il modello se l'ambiente è diverso.

@description('Location for cloud resources')
param mqExtensionName string = 'mq'
param clusterName string = 'clusterName'
param eventHubNamespaceName string = 'default'

resource connectedCluster 'Microsoft.Kubernetes/connectedClusters@2021-10-01' existing = {
  name: clusterName
}

resource mqExtension 'Microsoft.KubernetesConfiguration/extensions@2022-11-01' existing = {
  name: mqExtensionName
  scope: connectedCluster
}

resource ehNamespace 'Microsoft.EventHub/namespaces@2021-11-01' existing = {
  name: eventHubNamespaceName
}

// Role assignment for Event Hubs Data Receiver role
resource roleAssignmentDataReceiver 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(ehNamespace.id, mqExtension.id, '7f951dda-4ed3-4680-a7ca-43fe172d538d')
  scope: ehNamespace
  properties: {
     // ID for Event Hubs Data Receiver role is a638d3c7-ab3a-418d-83e6-5f17a39d4fde
    roleDefinitionId: resourceId('Microsoft.Authorization/roleDefinitions', 'a638d3c7-ab3a-418d-83e6-5f17a39d4fde') 
    principalId: mqExtension.identity.principalId
    principalType: 'ServicePrincipal'
  }
}

// Role assignment for Event Hubs Data Sender role
resource roleAssignmentDataSender 'Microsoft.Authorization/roleAssignments@2022-04-01' = {
  name: guid(ehNamespace.id, mqExtension.id, '69b88ce2-a752-421f-bd8b-e230189e1d63')
  scope: ehNamespace
  properties: {
    // ID for Event Hubs Data Sender role is 2b629674-e913-4c01-ae53-ef4638d8f975
    roleDefinitionId: resourceId('Microsoft.Authorization/roleDefinitions', '2b629674-e913-4c01-ae53-ef4638d8f975') 
    principalId: mqExtension.identity.principalId
    principalType: 'ServicePrincipal'
  }
}

# Set the required environment variables

# Resource group for resources
RESOURCE_GROUP=xxx

# Bicep template files name
TEMPLATE_FILE_NAME=xxx

# MQ Arc extension name
MQ_EXTENSION_NAME=xxx

# Arc connected cluster name
CLUSTER_NAME=xxx

# Event Hubs namespace name
EVENTHUB_NAMESPACE=xxx


az deployment group create \
      --name assign-RBAC-roles \
      --resource-group $RESOURCE_GROUP \
      --template-file $TEMPLATE_FILE_NAME \
      --parameters mqExtensionName=$MQ_EXTENSION_NAME \
      --parameters clusterName=$CLUSTER_NAME \
      --parameters eventHubNamespaceName=$EVENTHUB_NAMESPACE

Kafka Connessione or

La risorsa personalizzata Kafka Connessione or consente di configurare un connettore Kafka in grado di comunicare un host Kafka e Hub eventi. Il connettore Kafka può trasferire dati tra argomenti MQTT e argomenti Kafka, usando Hub eventi come endpoint compatibile con Kafka.

L'esempio seguente illustra un'istanza di Kafka Connessione or CR che si connette a un endpoint di Hub eventi usando l'identità di Azure di IoT MQ, presuppone che siano state installate altre risorse MQ usando la guida introduttiva:

apiVersion: mq.iotoperations.azure.com/v1beta1
kind: KafkaConnector
metadata:
  name: my-eh-connector
  namespace: azure-iot-operations # same as one used for other MQ resources
spec:
  image:
    pullPolicy: IfNotPresent
    repository: mcr.microsoft.com/azureiotoperations/kafka
    tag: 0.4.0-preview
  instances: 2
  clientIdPrefix: my-prefix
  kafkaConnection:
    # Port 9093 is Event Hubs' Kakfa endpoint
    # Plug in your Event Hubs namespace name
    endpoint: <NAMESPACE>.servicebus.windows.net:9093
    tls:
      tlsEnabled: true
    authentication:
      enabled: true
      authType:
        systemAssignedManagedIdentity:
          # plugin in your Event Hubs namespace name
          audience: "https://<NAMESPACE>.servicebus.windows.net" 
  localBrokerConnection:
    endpoint: "aio-mq-dmqtt-frontend:8883"
    tls:
      tlsEnabled: true
      trustedCaCertificateConfigMap: "aio-ca-trust-bundle-test-only"
    authentication:
      kubernetes: {}

La tabella seguente descrive i campi in Kafka Connessione or CR:

Campo	Descrizione	Richiesto
image	Immagine del connettore Kafka. È possibile specificare , pullPolicyrepositorye tag dell'immagine. I valori predefiniti vengono visualizzati nell'esempio precedente.	Sì
instances	Numero di istanze del connettore Kafka da eseguire.	Sì
clientIdPrefix	Stringa da anteporre a un ID client usato dal connettore.	No
kafka Connessione ion	Dettagli della connessione dell'endpoint di Hub eventi. Vedere Kafka Connessione ion.	Sì
localBroker Connessione ion	Dettagli della connessione del broker locale che esegue l'override della connessione broker predefinita. Vedere Gestire la connessione broker locale.	No
logLevel	Livello di log del connettore Kafka. I valori possibili sono: traccia, debug, info, avviso, errore o irreversibile. Il valore predefinito è warn.	No

Connessione Kafka

Il kafkaConnection campo definisce i dettagli della connessione dell'endpoint Kafka.

Campo	Descrizione	Richiesto
endpoint	Host e porta dell'endpoint di Hub eventi. La porta è in genere 9093. È possibile specificare più endpoint separati da virgole per usare la sintassi dei server bootstrap.	Sì
tls	Configurazione per la crittografia TLS. Vedere TLS.	Sì
autenticazione	Configurazione per l'autenticazione. Vedere Autenticazione.	No

TLS

Il tls campo abilita la crittografia TLS per la connessione e, facoltativamente, specifica una mappa di configurazione della CA.

Campo	Descrizione	Richiesto
tlsEnabled	Valore booleano che indica se la crittografia TLS è abilitata o meno. Deve essere impostato su true per la comunicazione di Hub eventi.	Sì
trustedCaCertificateConfigMap	Nome della mappa di configurazione contenente il certificato ca per la verifica dell'identità del server. Questo campo non è obbligatorio per la comunicazione di Hub eventi, perché Hub eventi usa ca note attendibili per impostazione predefinita. Tuttavia, è possibile usare questo campo se si vuole usare un certificato ca personalizzato.	No

Quando si specifica una CA attendibile è necessaria, creare un Oggetto ConfigMap contenente la pozione pubblica della CA in formato PEM e specificare il nome nella trustedCaCertificateConfigMap proprietà .

kubectl create configmap ca-pem --from-file path/to/ca.pem

Autenticazione

Il campo di autenticazione supporta diversi tipi di metodi di autenticazione, ad esempio SASL, X509 o identità gestita.

Campo	Descrizione	Richiesto
Enabled	Valore booleano che indica se l'autenticazione è abilitata o meno.	Sì
Authtype	Campo contenente il tipo di autenticazione utilizzato. Vedere Tipo di autenticazione

Tipo di autenticazione

Campo	Descrizione	Richiesto
sasl	Configurazione per l'autenticazione SASL. saslTypeSpecificare , che può essere semplice, scramSha256 o scramSha512 e token fare riferimento al segreto Kubernetes secretName o Azure Key Vault keyVault contenente la password.	Sì, se si usa l'autenticazione SASL
systemAssignedManagedIdentity	Configurazione per l'autenticazione dell'identità gestita. Specificare il gruppo di destinatari per la richiesta di token, che deve corrispondere allo spazio dei nomi di Hub eventi (https://<NAMESPACE>.servicebus.windows.net) perché il connettore è un client Kafka. Un'identità gestita assegnata dal sistema viene creata e assegnata automaticamente al connettore quando è abilitata.	Sì, se si usa l'autenticazione dell'identità gestita
x509	Configurazione per l'autenticazione X509. Specificare il secretName campo o keyVault . Il secretName campo è il nome del segreto che contiene il certificato client e la chiave client in formato PEM, archiviati come segreto TLS.	Sì, se si usa l'autenticazione X509

Per informazioni su come usare Azure Key Vault e keyVault per gestire i segreti per Azure IoT MQ invece dei segreti Kubernetes, vedere Gestire i segreti usando Azure Key Vault o i segreti Kubernetes.

Eseguire l'autenticazione in Hub eventi

Per connettersi a Hub eventi usando un stringa di connessione e un segreto Kubernetes, usare plain il tipo SASL e $ConnectionString come nome utente e il stringa di connessione completo come password. Creare prima di tutto il segreto Kubernetes:

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

Quindi, fare riferimento al segreto nella configurazione:

authentication:
  enabled: true
  authType:
    sasl:
      saslType: plain
      token:
        secretName: cs-secret

Per usare Azure Key Vault invece dei segreti Kubernetes, creare un segreto di Azure Key Vault con il stringa di connessione Endpoint=sb://.., farvi riferimento con vaultSecrete specificare il nome utente come "$ConnectionString" nella configurazione.

authentication:
  enabled: true
  authType:
    sasl:
      saslType: plain
      token:
        keyVault:
          username: "$ConnectionString"
          vault:
            name: my-key-vault
            directoryId: <AKV directory ID>
            credentials:
              servicePrincipalLocalSecretName: aio-akv-sp
          vaultSecret:
            name: my-cs # Endpoint=sb://..
            # version: 939ecc2...

Per usare l'identità gestita, specificarla come unico metodo con l'autenticazione. È anche necessario assegnare un ruolo all'identità gestita che concede l'autorizzazione per inviare e ricevere messaggi da Hub eventi, ad esempio Hub eventi di Azure Proprietario dati o Hub eventi di Azure mittente/ricevitore dati. Per altre informazioni, vedere Autenticare un'applicazione con Microsoft Entra ID per accedere alle risorse di Hub eventi.

authentication:
  enabled: true
  authType:
    systemAssignedManagedIdentity:
      audience: https://<NAMESPACE>.servicebus.windows.net

X.509

Per X.509, usare il segreto TLS kubernetes contenente il certificato pubblico e la chiave privata.

kubectl create secret tls my-tls-secret -n azure-iot-operations \
  --cert=path/to/cert/file \
  --key=path/to/key/file

Specificare quindi nella secretName configurazione.

authentication:
  enabled: true
  authType:
    x509:
      secretName: my-tls-secret

Per usare invece Azure Key Vault, assicurarsi che il certificato e la chiave privata siano importati correttamente e quindi specificare il riferimento con vaultCert.

authentication:
  enabled: true
  authType:
    x509:
      keyVault:
        vault:
          name: my-key-vault
          directoryId: <AKV directory ID>
          credentials:
            servicePrincipalLocalSecretName: aio-akv-sp
        vaultCert:
          name: my-cert
          # version: 939ecc2...
        ## If presenting full chain also  
        # vaultCaChainSecret:
        #   name: my-chain

In alternativa, se è necessaria la presentazione della catena completa, caricare il certificato completo della catena e la chiave in AKV come file PFX e usare invece il vaultCaChainSecret campo .

# ...
keyVault:
  vaultCaChainSecret:
    name: my-cert
    # version: 939ecc2...

Gestire la connessione broker locale

Come il bridge MQTT, il connettore hub eventi funge da client per il broker MQTT di IoT. Se è stata personalizzata la porta del listener e/o l'autenticazione del broker MQTT di IoT MQTT, eseguire l'override della configurazione della connessione MQTT locale anche per il connettore di Hub eventi. Per altre informazioni, vedere Connessione broker locale del bridge MQTT.

Kafka Connessione orTopicMap

La risorsa personalizzata Kafka Connessione orTopicMap consente di definire il mapping tra argomenti MQTT e argomenti Kafka per il trasferimento bidirezionale dei dati. Specificare un riferimento a kafka Connessione or CR e un elenco di route. Ogni route può essere una route DA MQTT a Kafka o da Kafka a MQTT. Ad esempio:

apiVersion: mq.iotoperations.azure.com/v1beta1
kind: KafkaConnectorTopicMap
metadata:
  name: my-eh-topic-map
  namespace: <SAME NAMESPACE AS BROKER> # For example "default"
spec:
  kafkaConnectorRef: my-eh-connector
  compression: snappy
  batching:
    enabled: true
    latencyMs: 1000
    maxMessages: 100
    maxBytes: 1024
  partitionStrategy: property
  partitionKeyProperty: device-id
  copyMqttProperties: true
  routes:
    # Subscribe from MQTT topic "temperature-alerts/#" and send to Kafka topic "receiving-event-hub"
    - mqttToKafka:
        name: "route1"
        mqttTopic: temperature-alerts/#
        kafkaTopic: receiving-event-hub
        kafkaAcks: one
        qos: 1
        sharedSubscription:
          groupName: group1
          groupMinimumShareNumber: 3
    # Pull from kafka topic "sending-event-hub" and publish to MQTT topic "heater-commands"
    - kafkaToMqtt:
        name: "route2"
        consumerGroupId: mqConnector
        kafkaTopic: sending-event-hub
        mqttTopic: heater-commands
        qos: 0

La tabella seguente descrive i campi in Kafka Connessione orTopicMap CR:

Campo	Descrizione	Richiesto
kafka Connessione orRef	Nome di Kafka Connessione or CR a cui appartiene la mappa di questo argomento.	Sì
compressione	Configurazione per la compressione dei messaggi inviati ad argomenti Kafka. Vedere Compressione.	No
divisione in batch	Configurazione per l'invio in batch dei messaggi inviati ad argomenti Kafka. Vedere Invio in batch.	No
partitionStrategy	Strategia per la gestione delle partizioni Kafka durante l'invio di messaggi agli argomenti Kafka. Vedere Strategia di gestione delle partizioni.	No
copyMqttProperties	Valore booleano per controllare se le proprietà del sistema e dell'utente MQTT vengono copiate nell'intestazione del messaggio Kafka. Le proprietà utente vengono copiate così come sono. Alcune trasformazioni vengono eseguite con le proprietà di sistema. Il valore predefinito è falso.	No
routes	Elenco di route per il trasferimento dei dati tra argomenti MQTT e argomenti Kafka. Ogni route può avere un mqttToKafka campo o kafkaToMqtt , a seconda della direzione del trasferimento dei dati. Vedere Route.	Sì

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.

Valore	Descrizione
Nessuno	Non viene applicata alcuna compressione o invio in batch. nessuno è il valore predefinito se non viene specificata alcuna compressione.
gzip	Vengono applicati la compressione e l'invio in batch GZIP. GZIP è un algoritmo di compressione per utilizzo generico che offre un buon equilibrio tra rapporto di compressione e velocità.
snappy	Vengono applicati la compressione snappy e l'invio in batch. Snappy è un algoritmo di compressione veloce che offre un rapporto di compressione moderato e velocità.
lz4	Vengono applicati la compressione e l'invio in batch LZ4. LZ4 è un algoritmo di compressione veloce che offre un rapporto di compressione basso e una velocità elevata.

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.

Campo	Descrizione	Richiesto
Enabled	Valore booleano che indica se l'invio in batch è abilitato o meno. Se non è impostato, il valore predefinito è false.	Sì
latencyMs	Intervallo di tempo massimo in millisecondi che i messaggi possono essere memorizzati nel buffer prima dell'invio. Se viene raggiunto questo intervallo, tutti i messaggi memorizzati nel buffer vengono inviati come batch, indipendentemente dal numero o dalla loro dimensione. Se non è impostato, il valore predefinito è 5.	No
maxMessages	Numero massimo di messaggi che possono essere memorizzati nel buffer prima dell'invio. Se questo numero viene raggiunto, tutti i messaggi memorizzati nel buffer vengono inviati come batch, indipendentemente dalla loro dimensione o dalla durata del buffer. Se non è impostato, il valore predefinito è 100000.	No
maxBytes	Dimensioni massime in byte che possono essere memorizzate nel buffer prima dell'invio. Se vengono raggiunte queste dimensioni, tutti i messaggi memorizzati nel buffer vengono inviati come batch, indipendentemente dal numero o dalla durata del buffer. Il valore predefinito è 1000000 (1 MB).	No

Un esempio di utilizzo dell'invio in batch è:

batching:
  enabled: true
  latencyMs: 1000
  maxMessages: 100
  maxBytes: 1024

Ciò significa che i messaggi vengono inviati quando sono presenti 100 messaggi nel buffer, o quando sono presenti 1024 byte nel buffer o quando sono trascorsi 1000 millisecondi dall'ultimo invio, a ogni prima occorrenza.

Strategia di gestione delle partizioni

La strategia di gestione delle partizioni è una funzionalità che consente di controllare il modo in cui i messaggi vengono assegnati alle partizioni Kafka durante l'invio ad 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 usati per identificare e ordinare i messaggi.

Per impostazione predefinita, il connettore Kafka 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.

Valore	Descrizione
impostazione predefinita	Assegna messaggi a partizioni casuali, usando un algoritmo round robin. È il valore predefinito se non viene specificata alcuna strategia.
static	Assegna messaggi a un numero di partizione fisso derivato dall'ID istanza del connettore. Ciò significa che ogni istanza del connettore invia messaggi a una partizione diversa. Ciò consente di ottenere un migliore bilanciamento del carico e località dei dati.
argomento	Usa il nome dell'argomento MQTT come chiave per il partizionamento. Ciò significa che i messaggi con lo stesso nome dell'argomento MQTT vengono inviati alla stessa partizione. Ciò consente di ottenere un ordinamento dei messaggi e una località dei dati migliori.
proprietà	Usa una proprietà del messaggio MQTT come chiave per il partizionamento. Specificare il nome della proprietà nel partitionKeyProperty campo . Ciò significa che i messaggi con lo stesso valore della proprietà vengono inviati alla stessa partizione. Ciò consente di ottenere un ordinamento dei messaggi e una località dei dati migliori in base a un criterio personalizzato.

Un esempio di utilizzo della strategia di gestione delle partizioni è:

partitionStrategy: property
partitionKeyProperty: device-id

Ciò significa che i messaggi con la stessa proprietà device-id vengono inviati alla stessa partizione.

Route

Il campo route definisce un elenco di route per il trasferimento dei dati tra argomenti MQTT e argomenti Kafka. Ogni route può avere un mqttToKafka campo o kafkaToMqtt , a seconda della direzione del trasferimento dei dati.

DA MQTT a Kafka

Il mqttToKafka campo definisce una route che trasferisce i dati da un argomento MQTT a un argomento Kafka.

Campo	Descrizione	Richiesto
name	Nome univoco per la route.	Sì
mqttTopic	Argomento MQTT da cui eseguire la sottoscrizione. È possibile usare caratteri jolly (# e +) per trovare le corrispondenze con più argomenti.	Sì
kafkaTopic	Argomento Kafka a cui inviare.	Sì
kafkaAcks	Il numero di riconoscimenti richiesti dal connettore dall'endpoint Kafka. I valori possibili sono: zero , oneo all.	No
qos	Livello di qualità del servizio (QoS) per la sottoscrizione dell'argomento MQTT. I valori possibili sono: 0 o 1 (impostazione predefinita). QoS 2 non è attualmente supportato.	Sì
sharedSubscription	Configurazione per l'uso di sottoscrizioni condivise per gli argomenti MQTT. groupNameSpecificare , che è un identificatore univoco per un gruppo di sottoscrittori e groupMinimumShareNumber, ovvero il numero di sottoscrittori in un gruppo che riceve messaggi da un argomento. Ad esempio, se groupName è "group1" e groupMinimumShareNumber è 3, il connettore crea tre sottoscrittori con lo stesso nome di gruppo per ricevere messaggi da un argomento. Questa funzionalità consente di distribuire messaggi tra più sottoscrittori senza perdere messaggi o creare duplicati.	No

Esempio di utilizzo mqttToKafka della route:

mqttToKafka:
  mqttTopic: temperature-alerts/#
  kafkaTopic: receiving-event-hub
  kafkaAcks: one
  qos: 1
  sharedSubscription:
    groupName: group1
    groupMinimumShareNumber: 3

In questo esempio, i messaggi degli argomenti MQTT che corrispondono a temperature-alerts/# vengono inviati all'argomento Kafka receiving-event-hub con il gruppo di sottoscrizioni QoS equivalente a 1 e condiviso "group1" con il numero di condivisione 3.

Da Kafka a MQTT

Il kafkaToMqtt campo definisce una route che trasferisce i dati da un argomento Kafka a un argomento MQTT.

Campo	Descrizione	Richiesto
name	Nome univoco per la route.	Sì
kafkaTopic	Argomento Kafka da cui eseguire il pull.	Sì
mqttTopic	Argomento MQTT in cui eseguire la pubblicazione.	Sì
consumerGroupId	Prefisso dell'ID del gruppo di consumer per ogni route Da Kafka a MQTT. Se non è impostato, l'ID del gruppo di consumer viene impostato sullo stesso nome della route.	No
qos	Livello di qualità del servizio (QoS) per i messaggi pubblicati nell'argomento MQTT. I valori possibili sono 0 o 1 (impostazione predefinita). QoS 2 non è attualmente supportato. Se QoS è impostato su 1, il connettore pubblica il messaggio nell'argomento MQTT e quindi attende l'ack prima di eseguire il commit del messaggio in Kafka. Per QoS 0, il connettore esegue il commit immediatamente senza l'ack MQTT.	No

Esempio di utilizzo kafkaToMqtt della route:

kafkaToMqtt:
  kafkaTopic: sending-event-hub
  mqttTopic: heater-commands
  qos: 0

In questo esempio, i messaggi dell'argomento Kafka sending-event-hub* vengono pubblicati in comandi di riscaldamento dell'argomentoMQTT con livello QoS 0.

Il nome dell'hub eventi deve corrispondere all'argomento Kafka

Ogni singolo hub eventi non deve essere denominato esattamente come l'argomento Kafka previsto specificato nelle route. Inoltre, il stringa di connessione EntityPath deve corrispondere se stringa di connessione ha come ambito un hub eventi. Questo requisito è dovuto al fatto che lo spazio dei nomi di Hub eventi è analogo al cluster Kafka e il nome dell'hub eventi è analogo a un argomento Kafka, pertanto il nome dell'argomento Kafka deve corrispondere al nome dell'hub eventi.

Offset del gruppo di consumer Kafka

Se il connettore si disconnette o viene rimosso e reinstallato con lo stesso ID gruppo di consumer Kafka, l'offset del gruppo di consumer (l'ultima posizione da cui vengono letti i messaggi del consumer Kafka) viene archiviato in Hub eventi di Azure. Per altre informazioni, vedere Gruppo di consumer di Hub eventi e gruppo di consumer Kafka.

Versione MQTT

Questo connettore usa solo MQTT v5.

Contenuto correlato

Pubblicare e sottoscrivere messaggi MQTT con l'anteprima mq di Azure IoT