Configure data flow endpoints for Microsoft Fabric Real-Time Intelligence

This method uses a managed identity to authenticate with the event stream. Use either system-assigned managed identity or user-assigned managed identity when you configure the data flow endpoint.

1. The connection details are in the Fabric portal under the Sources section of your event stream.

2. In the details panel for the custom endpoint, select Kafka protocol.

3. Select the Entra ID Authentication section to view the connection details.

4. Copy the details for the values for the Bootstrap server and Topic name. You use these values to configure the data flow endpoint.

   

   Settings	Description
   Bootstrap server	The bootstrap server address is used for the hostname property in data flow endpoint.
   Topic name	The event hub name is used as the Kafka topic and is in the format es_aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb.

1. The connection details are in the Fabric portal under the Sources section of your event stream.

2. In the details panel for the custom endpoint, select Kafka protocol.

3. Select the SAS Key Authentication section to view the connection details.

4. Copy the details for the values for the Bootstrap server, Topic name, and Connection string-primary key. You use these values to configure the data flow endpoint.

   

   Settings	Description
   Bootstrap server	The bootstrap server address is used for the hostname property in data flow endpoint.
   Topic name	The event hub name is used as the Kafka topic and is in the format es_aaaaaaaa-0000-1111-2222-bbbbbbbbbbbb.
   Connection string-primary key	The connection string with the primary key.

1. In the IoT Operations experience portal, select the Data flow endpoints tab.

2. Under Create new data flow endpoint, select Microsoft Fabric Real-Time Intelligence > New.

3. Enter the following settings for the endpoint.

   

   Setting	Description
   Name	The name of the data flow endpoint.
   Host	The hostname of the event stream custom endpoint in the format <bootstrap-server>.servicebus.windows.net:9093. Use the bootstrap server address noted previously.
   Authentication method	The method used for authentication. Choose System assigned managed identity, User assigned managed identity, or SASL.

   Use the authentication method that matches how you want Azure IoT Operations to connect to the Fabric event stream custom endpoint:

   • System-assigned managed identity: Use the Azure IoT Operations Arc extension identity to authenticate with the event stream. Before you create the endpoint, add the extension identity to the Fabric workspace with Contributor or higher permissions. To learn more, see System-assigned managed identity.
   • User-assigned managed identity: Use a user-assigned managed identity configured for Azure IoT Operations cloud connections. Before you create the endpoint, add the user-assigned managed identity to the Fabric workspace with Contributor or higher permissions. To learn more, see User-assigned managed identity.
   • SASL: Use the Fabric event stream custom endpoint connection string. This method requires SASL settings and a synced Kubernetes secret with the username and password values. To learn more, see SASL.

4. Select Apply to provision the endpoint.

Create or replace

Use the az iot ops dataflow endpoint create fabric-realtime command to create or replace a Microsoft Fabric Real-Time Intelligence data flow endpoint.

az iot ops dataflow endpoint create fabric-realtime --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <EndpointName> --host "<BootstrapServerAddress>"

Here's an example command to create or replace a Microsoft Fabric Real-Time Intelligence data flow endpoint named fabric-realtime-endpoint:

az iot ops dataflow endpoint create fabric-realtime --resource-group myResourceGroup --instance myAioInstance --name fabric-realtime-endpoint --host "fabricrealtime.servicebus.windows.net:9093"

Create or change

Use the az iot ops dataflow endpoint apply command to create or change a Microsoft Fabric Real-Time Intelligence data flow endpoint.

az iot ops dataflow endpoint apply --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <EndpointName> --config-file <ConfigFilePathAndName>

The --config-file parameter is the path and file name of a JSON configuration file containing the resource properties.

In this example, assume a configuration file named fabric-realtime-endpoint.json with the following content stored in the user's home directory:

{
  "endpointType": "Kafka",
  "kafkaSettings": {
    "host": "<BootstrapServerAddress>",
    "authentication": {
      "method": "Sasl",
      "saslSettings": {
        "saslType": "Plain",
        "secretRef": "<SecretName>"
      }
    },
    "tls": {
      "mode": "Enabled"
    }
  }
}

Here's an example command to create a new Microsoft Fabric Real-Time Intelligence data flow endpoint named fabric-realtime-endpoint:

az iot ops dataflow endpoint apply --resource-group myResourceGroupName --instance myAioInstanceName --name fabric-realtime-endpoint --config-file ~/fabric-realtime-endpoint.json

Create a Bicep .bicep file with the following content.

param aioInstanceName string = '<AIO_INSTANCE_NAME>'
param customLocationName string = '<CUSTOM_LOCATION_NAME>'
param endpointName string = '<ENDPOINT_NAME>'
param hostName string = '<BOOTSTRAP_SERVER_ADDRESS>'
param secretName string = '<SECRET_NAME>'

resource customLocation 'Microsoft.ExtendedLocation/customLocations@2021-08-31-preview' existing = {
  name: customLocationName
}

resource aioInstance 'Microsoft.IoTOperations/instances@2024-11-01' existing = {
  name: aioInstanceName
}

resource fabricRealtimeEndpoint 'Microsoft.IoTOperations/instances/dataflowEndpoints@2024-11-01' = {
  parent: aioInstance
  name: endpointName
  extendedLocation: {
    name: customLocation.id
    type: 'CustomLocation'
  }
  properties: {
    endpointType: 'Kafka'
    kafkaSettings: {
      host: hostName
      authentication: {
        method: 'Sasl'
        saslSettings: {
          saslType: 'Plain' // Or ScramSha256, ScramSha512
          secretRef: secretName
        }
      }
      tls: {
        mode: 'Enabled'
      }
    }
  }
}

Then, deploy via Azure CLI.

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

kubectl create secret generic <SECRET_NAME> -n azure-iot-operations \
  --from-literal=username='$ConnectionString' \
  --from-literal=password='<ConnectionStringPrimaryKey>'

Create a Kubernetes manifest .yaml file with the following content.

apiVersion: connectivity.iotoperations.azure.com/v1
kind: DataflowEndpoint
metadata:
  name: <ENDPOINT_NAME>
  namespace: azure-iot-operations
spec:
  endpointType: Kafka
  kafkaSettings:
    host: <BOOTSTRAP_SERVER_ADDRESS>
    authentication:
      method: Sasl
      saslSettings:
        saslType: Plain # Or ScramSha256, ScramSha512
        secretRef: <SECRET_NAME>
    tls:
      mode: Enabled

Then apply the manifest file to the Kubernetes cluster.

kubectl apply -f <FILE>.yaml

In the operations experience data flow endpoint settings page, select the Basic tab then choose Authentication method > System assigned managed identity.

Create or replace

Use the az iot ops dataflow endpoint create fabric-realtime command with the --auth-type parameter set to SystemAssignedManagedIdentity for system-assigned managed identity authentication. In most cases, you don't need to specify --audience. The default audience matches the endpoint host in the form https://<NAMESPACE>.servicebus.windows.net.

az iot ops dataflow endpoint create fabric-realtime --auth-type SystemAssignedManagedIdentity --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <EndpointName> --host "<BootstrapServerAddress>"

Create or change

Use the az iot ops dataflow endpoint apply command with the --config-file parameter.

In this example, assume a configuration file with the following content:

{
    "endpointType": "Kafka",
    "kafkaSettings": {
        "host": "fabricrealtime.servicebus.windows.net:9093",
        "authentication": {
            "method": "SystemAssignedManagedIdentity",
            "systemAssignedManagedIdentitySettings": {}
        },
        "tls": {
            "mode": "Enabled"
        }
    }
}

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

kafkaSettings:
  authentication:
    method: SystemAssignedManagedIdentity
    systemAssignedManagedIdentitySettings:
      {}

In the operations experience data flow endpoint settings page, select the Basic tab then choose Authentication method > User assigned managed identity.

Create or replace

Use the az iot ops dataflow endpoint create command with the --auth-type parameter set to UserAssignedManagedIdentity for with user-assigned managed identity authentication.

az iot ops dataflow endpoint create fabric-realtime --auth-type UserAssignedManagedIdentity --client-id <ClientId> --tenant-id <TenantId> --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <EndpointName> --host "<BootstrapServerAddress>"

Create or change

Use the az iot ops dataflow endpoint apply with the --config-file parameter

In this example, assume a configuration file with the following content:

{
    "endpointType": "Kafka",
    "kafkaSettings": {
        "host": "<BootstrapServerAddress>",
        "authentication": {
          "method": "UserAssignedManagedIdentity",
          "userAssignedManagedIdentitySettings": {
            "clientId": "<ID>",
            "tenantId": "<ID>"
          }
        },
        "tls": {
            "mode": "Enabled"
        }
    }
}

The scope property is optional. If needed, set it to the audience for the Kafka broker, such as https://<BootstrapServerAddress>.

kafkaSettings: {
  authentication: {
    method: 'UserAssignedManagedIdentity'
    userAssignedManagedIdentitySettings: {
      clientId: '<CLIENT_ID>'
      tenantId: '<TENANT_ID>'
      // Optional
      // scope: 'https://<SCOPE_URL>'
    }
  }
  ...
}

kafkaSettings:
  authentication:
    method: UserAssignedManagedIdentity
    userAssignedManagedIdentitySettings:
      clientId: <CLIENT_ID>
      tenantId: <TENANT_ID>
      # Optional
      # scope: https://<SCOPE_URL>

In the operations experience data flow endpoint settings page, select the Basic tab then choose Authentication method > SASL.

Enter the following settings for the endpoint:

Select Add reference to create a new or choose an existing Key Vault reference for the username and password references.

For Username reference of token secret, store a Key Vault secret whose value is the literal string $ConnectionString (including the leading $).

For Password reference of token secret, the secret value must be the connection string with the primary key from the event stream custom endpoint.

Create or replace

Use the az iot ops dataflow endpoint create command with the --auth-type parameter set to Sasl for SASL authentication.

az iot ops dataflow endpoint create fabric-realtime --auth-type Sasl --sasl-type <SaslType> --secret-name <SecretName> --resource-group <ResourceGroupName> --instance <AioInstanceName> --name <EndpointName> --host "<BootstrapServerAddress>"

Create or change

Use the az iot ops dataflow endpoint apply with the --config-file parameter

In this example, assume a configuration file with the following content:

{
    "endpointType": "Kafka",
    "kafkaSettings": {
        "host": "<BootstrapServerAddress>",
        "authentication": {
          "method": "Sasl",
          "saslSettings": {
            "saslType": "<SaslType>",
            "secretRef": "<SecretName>"
          }
        },
        "tls": {
            "mode": "Enabled"
        }
    }
}

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

kubectl create secret generic <SECRET_NAME> -n azure-iot-operations \
  --from-literal=username='$ConnectionString' \
  --from-literal=password='<ConnectionStringPrimaryKey>'

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