eventstream-consumption-cli

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}" \
  --resource "https://api.fabric.microsoft.com"

az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/getDefinition" \
  --resource "https://api.fabric.microsoft.com" \
  --body '{}'

# Using jq + base64 (Linux; on macOS use base64 -D instead of -d)
az rest --method POST \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/getDefinition" \
  --resource "https://api.fabric.microsoft.com" \
  --body '{}' \
  | jq -r '.definition.parts[] | select(.path=="eventstream.json") | .payload' \
  | base64 -d | jq .

# PowerShell (Windows)
$def = az rest --method POST `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/getDefinition" `
  --resource "https://api.fabric.microsoft.com" `
  --body '{}' | ConvertFrom-Json
$payload = ($def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' }).payload
[Text.Encoding]::UTF8.GetString([Convert]::FromBase64String($payload)) | ConvertFrom-Json | ConvertTo-Json -Depth 10

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/topology" \
  --resource "https://api.fabric.microsoft.com"

# Extract the sourceId for a Custom Endpoint source (use name filter if multiple exist)
SOURCE_ID=$(az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/topology" \
  --resource "https://api.fabric.microsoft.com" \
  | jq -r '[.sources[] | select(.type=="CustomEndpoint")] | if length == 0 then error("No Custom Endpoint sources found in this Eventstream") elif length > 1 then error("Multiple Custom Endpoint sources found — filter by .name") else .[0].id end') \
  || { echo "Failed to resolve Custom Endpoint source ID"; exit 1; }
 
if [ -z "$SOURCE_ID" ]; then echo "SOURCE_ID is empty — check topology output"; exit 1; fi

# PowerShell — extract sourceId for Custom Endpoint (fails clearly if multiple exist)
$topology = az rest --method GET `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
$customSources = @($topology.sources | Where-Object { $_.type -eq 'CustomEndpoint' })
if ($customSources.Count -eq 0) { throw "No Custom Endpoint sources found in this Eventstream" }
if ($customSources.Count -gt 1) { throw "Multiple Custom Endpoint sources found. Filter by name: $($customSources.name -join ', ')" }
$sourceId = $customSources[0].id

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams/${EVENTSTREAM_ID}/sources/${SOURCE_ID}/connection" \
  --resource "https://api.fabric.microsoft.com"

az rest --method GET `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$WORKSPACE_ID/eventstreams/$EVENTSTREAM_ID/sources/$sourceId/connection" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json

{
  "fullyQualifiedNamespace": "namespace.servicebus.windows.net",
  "eventHubName": "es_<guid>",
  "accessKeys": {
    "primaryKey": "...",
    "secondaryKey": "...",
    "primaryConnectionString": "Endpoint=sb://namespace.servicebus.windows.net/;...",
    "secondaryConnectionString": "..."
  }
}

# 1. Discover workspace ID
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }
 
# 2. List Eventstreams (handles pagination)
$allItems = @()
$resp = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
$allItems += $resp.value
while ($resp.continuationUri) {
    $resp = az rest --method get `
      --url $resp.continuationUri `
      --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
    $allItems += $resp.value
}
$allItems | Select-Object displayName, id, description | Format-Table

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }
 
$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }
 
# 2. Get topology (returns JSON directly — no base64 decoding needed)
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
 
# 3. Summarize nodes (filter nulls from arrays)
Write-Host "Sources:"
@($topo.sources | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type), id: $($_.id))"
}
Write-Host "Operators:"
@($topo.operators | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}
Write-Host "Destinations:"
@($topo.destinations | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}
Write-Host "Streams:"
@($topo.streams | Where-Object { $_ -ne $null }) | ForEach-Object {
    Write-Host "  - $($_.name) (type: $($_.type))"
}

# 1. Discover workspace + Eventstream IDs
$wsId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='analytics'] | [0].id" -o tsv)
if (-not $wsId) { throw "Workspace 'analytics' not found" }
 
$esId = (az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/items?type=Eventstream" `
  --resource "https://api.fabric.microsoft.com" `
  --query "value[?displayName=='SensorIngestion'] | [0].id" -o tsv)
if (-not $esId) { throw "Eventstream 'SensorIngestion' not found" }
 
# 2. Get definition (handles LRO if API returns 202)
$respJson = az rest --method post `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/getDefinition" `
  --resource "https://api.fabric.microsoft.com" `
  --headers "Content-Type=application/json" `
  --body '{}'
if ($LASTEXITCODE -ne 0 -or -not $respJson) { throw "getDefinition request failed" }
$resp = $respJson | ConvertFrom-Json
 
if ($resp.definition) {
    $def = $resp  # Synchronous — got definition immediately
} elseif ($resp.operationId) {
    # Asynchronous (LRO) — poll operation status
    $def = $null
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds 5
        $status = (az rest --method get `
          --url "https://api.fabric.microsoft.com/v1/operations/$($resp.operationId)" `
          --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json)
        if ($status.status -eq 'Succeeded') {
            $def = (az rest --method get `
              --url "https://api.fabric.microsoft.com/v1/operations/$($resp.operationId)/result" `
              --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json)
            break
        } elseif ($status.status -in @('Failed', 'Cancelled')) {
            throw "getDefinition LRO $($status.status): $($status.error.message)"
        }
    }
    if (-not $def.definition) { throw "getDefinition LRO timed out (last status: $($status.status))" }
} else {
    throw "getDefinition returned neither a definition nor an operationId"
}
 
# 3. Decode eventstreamProperties.json part (holds retention + throughput)
$propsPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstreamProperties.json' } | Select-Object -First 1
if ($propsPart) {
    $props = [Text.Encoding]::UTF8.GetString(
      [Convert]::FromBase64String($propsPart.payload)) | ConvertFrom-Json
    Write-Host "Retention: $($props.retentionTimeInDays) days"
    Write-Host "Throughput Level: $($props.eventThroughputLevel)"
} else {
    # Fall back to topology-level properties (older format)
    $esPart = $def.definition.parts | Where-Object { $_.path -eq 'eventstream.json' } | Select-Object -First 1
    if (-not $esPart) { throw "eventstream.json part not found in definition" }
    $topology = [Text.Encoding]::UTF8.GetString(
      [Convert]::FromBase64String($esPart.payload)) | ConvertFrom-Json
    if ($topology.properties.retentionTimeInDays) {
        Write-Host "Retention: $($topology.properties.retentionTimeInDays) days"
    } else {
        Write-Host "Retention: (platform default — not explicitly configured)"
    }
    if ($topology.properties.eventThroughputLevel) {
        Write-Host "Throughput Level: $($topology.properties.eventThroughputLevel)"
    } else {
        Write-Host "Throughput Level: (platform default — not explicitly configured)"
    }
}

# 1. Discover workspace + Eventstream IDs (omitted for brevity)
 
# 2. Get topology to find Custom Endpoint source ID
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
 
$ceSource = @($topo.sources | Where-Object { $_.type -eq 'CustomEndpoint' }) | Select-Object -First 1
if (-not $ceSource) { throw "No CustomEndpoint source found in this Eventstream" }
$sourceId = $ceSource.id
 
# 3. Get connection metadata
$conn = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/sources/$sourceId/connection" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
 
Write-Host "Fully Qualified Namespace: $($conn.fullyQualifiedNamespace)"
Write-Host "Event Hub Name:            $($conn.eventHubName)"
Write-Host "Kafka Bootstrap Server:    $($conn.fullyQualifiedNamespace):9093"

# 1. Discover workspace + Eventstream IDs (omitted for brevity)
 
# 2. Get topology
$topo = az rest --method get `
  --url "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/topology" `
  --resource "https://api.fabric.microsoft.com" | ConvertFrom-Json
 
# 3. Validate sources
@($topo.sources | Where-Object { $_ -ne $null }) | ForEach-Object {
    $src = $_
    Write-Host "Source: $($src.name) (type: $($src.type))"
    if ($src.properties.dataConnectionId) {
        Write-Host "  Cloud connection: $($src.properties.dataConnectionId)"
    }
    if ($src.type -in @('AzureEventHub','AzureEventHubExtended','AzureIoTHub','ConfluentCloud','ApacheKafka','AmazonMSKKafka') -and
        -not $src.properties.consumerGroupName) {
        Write-Host "  WARNING: No consumer group set (required for $($src.type))"
    }
}
 
# 4. Validate destinations
@($topo.destinations | Where-Object { $_ -ne $null }) | ForEach-Object {
    $dst = $_
    Write-Host "Destination: $($dst.name) (type: $($dst.type))"
    if (-not $dst.inputNodes -or $dst.inputNodes.Count -eq 0) {
        Write-Host "  WARNING: No input wired — destination will receive no events"
    }
    if ($dst.type -eq 'Eventhouse' -and -not $dst.properties.tableName) {
        Write-Host "  WARNING: No target table configured"
    }
    if ($dst.type -eq 'Eventhouse' -and $dst.properties.dataIngestionMode -eq 'DirectIngestion') {
        if (-not $dst.properties.connectionName -or -not $dst.properties.mappingRuleName) {
            Write-Host "  WARNING: DirectIngestion requires connectionName and mappingRuleName"
        }
    }
}
 
# 5. Check node count limits
$ceCount = @($topo.sources | Where-Object { $_.type -eq 'CustomEndpoint' }).Count +
           @($topo.destinations | Where-Object { $_.type -eq 'CustomEndpoint' }).Count +
           @($topo.destinations | Where-Object { $_.type -eq 'Eventhouse' -and
             $_.properties.dataIngestionMode -eq 'DirectIngestion' }).Count
Write-Host "CustomEndpoint + DI count: $ceCount / 11 limit"
if ($ceCount -gt 11) {
    Write-Host "WARNING: Exceeds limit of 11 CustomEndpoint + DirectIngestion nodes"
}