eventstream-consumption-cli

アップデートの確認 — セッションごとに1回（必須） このスキルをセッションで初めて使用する際は、続行する前に check-updates スキルを実行してください。

• GitHub Copilot CLI / VS Code：check-updates スキルを呼び出します。
• Claude Code / Cowork / Cursor / Windsurf / Codex：ローカルとリモートの package.json のバージョンを比較します。
• このセッション内ですでに確認済みの場合はスキップしてください。

重要な注意事項

1. ワークスペース名から詳細（ID を含む）を確認するには、すべてのワークスペースを一覧表示してから、JMESPath でフィルタリングします。
2. ワークスペース ID、アイテムの種類、アイテム名からアイテムの詳細（ID を含む）を確認するには、そのワークスペース内にある該当する種類のすべてのアイテムを一覧表示してから、JMESPath でフィルタリングします。
3. Eventstream ≠ Eventhouse です。Eventstream は、リアルタイムのイベント取り込みとルーティングを行うパイプラインです。KQL クエリには eventhouse-consumption-cli を使用してください。

Eventstream Consumption — CLI スキル

目次

タスク	リファレンス	備考
Fabric のワークスペースとアイテムを見つける	COMMON-CLI.md § Fabric のワークスペースとアイテムを見つける	必須 — 最初にリンクを読む［名前からワークスペース ID を、名前、アイテムの種類、ワークスペース ID からアイテム ID を見つけるために必要です］
Fabric のトポロジと主要な概念	COMMON-CORE.md § Fabric のトポロジと主要な概念
環境 URL	COMMON-CORE.md § 環境 URL
認証とトークンの取得	COMMON-CORE.md § 認証とトークンの取得	対象ユーザーが間違っていると 401 になります。認証の問題が発生した場合は、まずお読みください
コアのコントロールプレーン REST API	COMMON-CORE.md § コアのコントロールプレーン REST API	ページネーション、LRO のポーリング、レート制限のパターンを含みます
注意事項、ベストプラクティス、トラブルシューティング	COMMON-CORE.md § 注意事項、ベストプラクティス、トラブルシューティング
ツール選択の理由	COMMON-CLI.md § ツール選択の理由
認証のレシピ	COMMON-CLI.md § 認証のレシピ	az login のフローとトークンの取得
az rest による Fabric コントロールプレーン API	COMMON-CLI.md § az rest による Fabric コントロールプレーン API	必ず --resource を指定してください。ページネーションと LRO ヘルパーを含みます
注意事項とトラブルシューティング（CLI 固有）	COMMON-CLI.md § 注意事項とトラブルシューティング（CLI 固有）	az rest の対象ユーザー、シェルのエスケープ、トークンの有効期限
クイックリファレンス	COMMON-CLI.md § クイックリファレンス	az rest のテンプレートとトークン対象ユーザー／ツールのマトリックス
Eventstream の一覧表示と検出	EVENTSTREAM-CONSUMPTION-CORE.md § Eventstream の一覧表示と検出	ワークスペースをまたいだ一覧表示、取得、検索
Eventstream のトポロジの確認	EVENTSTREAM-CONSUMPTION-CORE.md § Eventstream のトポロジの確認	base64 定義をデコードして、グラフのフローを追跡します
Eventstream の正常性の監視	EVENTSTREAM-CONSUMPTION-CORE.md § Eventstream の正常性の監視	保持期間とスループットを確認します
ソースと宛先の状態	EVENTSTREAM-CONSUMPTION-CORE.md § ソースと宛先の状態	ソースと宛先の検証チェックリスト
下流分析との統合	EVENTSTREAM-CONSUMPTION-CORE.md § 下流分析との統合	Eventhouse、Lakehouse、Activator、Real-Time Hub
注意事項とトラブルシューティングのリファレンス	EVENTSTREAM-CONSUMPTION-CORE.md § 注意事項とトラブルシューティングのリファレンス	よくある 10 件の問題と原因、解決方法
Eventstream の一覧表示	SKILL.md § Eventstream の一覧表示
Eventstream のトポロジの確認	SKILL.md § Eventstream のトポロジの確認	グラフをデコードして調べます
カスタムエンドポイントの接続文字列の取得	SKILL.md § カスタムエンドポイントの接続文字列の取得	Topology API を通じて Kafka/EH 接続を取得します
Eventstream の構成の検証	SKILL.md § Eventstream の構成の検証
注意事項、ルール、トラブルシューティング	SKILL.md § 注意事項、ルール、トラブルシューティング	必ず行う／避ける／優先するためのチェックリスト

---

Eventstream の一覧表示

ワークスペース内のすべての Eventstream を一覧表示する

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

Eventstream アイテムの配列を返します。JMESPath を使用して名前で絞り込みます。

az rest --method GET \
  --url "https://api.fabric.microsoft.com/v1/workspaces/${WORKSPACE_ID}/eventstreams" \
  --resource "https://api.fabric.microsoft.com" \
  --query "value[?displayName=='my-eventstream']"

Eventstream の詳細を取得する

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

---

Eventstream のトポロジを調べる

ヒント：Topology API（GET .../eventstreams/{id}/topology）は、ランタイムの状態、エラー情報、ノード ID を base64 デコードなしで返します。運用上の調査（ヘルスチェック、接続情報の取得）には、こちらを優先してください。トポロジの変更に必要な、作成時点のグラフ構造全体を取得する場合は、POST .../getDefinition（以下）を使用します。

Eventstream の定義を取得し、デコードしてグラフのトポロジ全体を調べます。

手順 1：定義を取得する

API に関する注意：Eventstream Definition API では、GET .../definition ではなく POST .../getDefinition を使用します。これは Fabric Items Definition のパターンに準拠しています。詳しくは公式ドキュメントをご覧ください。

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 '{}'

手順 2：トポロジをデコードする

eventstream.json 部分の payload フィールドを抽出し、base64 デコードします。

# 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

手順 3：トポロジを要約する

デコード後、各ノードタイプの数を数え、一覧表示します。

指標	デコード済み JSON 内のパス
ソース	.sources[] | .name, .type
宛先	.destinations[] | .name, .type
オペレーター	.operators[] | .name, .type
ストリーム	.streams[] | .name, .type

---

カスタムエンドポイントの接続文字列を取得する

POST .../getDefinition エンドポイントは、カスタムエンドポイントソースに対して空のプロパティを返します。Kafka／Event Hub の接続情報を取得するには、Topology API の /connection エンドポイントを使用します。

重要：このエンドポイントには Eventstream.ReadWrite.All のアクセス許可スコープが必要です（読み取り専用ではありません）。

手順 1：トポロジを取得してソース 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"

レスポンスからカスタムエンドポイントのソースノードを見つけ、その id を抽出します。

# 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

手順 2：接続の詳細を取得する

⚠️ セキュリティ：このエンドポイントはアクセスキーと接続文字列を返します。呼び出す前に、ユーザーから明示的な確認を得てください。ユーザーが安全なコンテキストでシークレット値を明示的に求めていない限り、表示する出力から primaryKey、secondaryKey、primaryConnectionString、secondaryConnectionString をマスキングしてください。認証情報をそのままログに記録しないでください。安全に保管し、必要に応じてローテーションしてください。

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": "..."
  }
}

Kafka プロデューサーの設定

レスポンスを使用して Kafka プロデューサーを設定します。

設定	値
bootstrap_servers	{fullyQualifiedNamespace}:9093
topic	{eventHubName}
security_protocol	SASL_SSL
sasl_mechanism	PLAIN
sasl_plain_username	$ConnectionString（固定のリテラル。変数ではありません）
sasl_plain_password	{primaryConnectionString}

制限事項：/connection エンドポイントがサポートされるのはカスタムエンドポイントソースのみです（Kafka／Event Hub の認証情報を返します）。その他のソースタイプ（Event Hub、IoT Hub など）では、接続設定（例：dataConnectionId、consumerGroup）はデコード済み定義のプロパティに直接保存されます。

---

Eventstream の設定を検証する

デコード済みの Eventstream トポロジの主要な設定項目を確認します。

ソース検証チェックリスト

確認項目	方法
ソースタイプが API でサポートされている	既知の 25 種類のタイプ列挙値と照合する
クラウド接続が存在する	dataConnectionId の GUID が解決できることを確認する
コンシューマーグループが設定されている	Event Hub、IoT Hub、Kafka ソースで必須
シリアル化形式がソースと一致している	inputSerialization.type が Json、Csv、または Avro であることを確認する

宛先の検証チェックリスト

チェック項目	方法
宛先の種類が有効	Lakehouse、Eventhouse、Activator、または CustomEndpoint である必要があります
対象アイテムにアクセス可能	workspaceId と itemId の組み合わせで GET が成功することを確認します
入力が接続されている	inputNodes 配列が空でない必要があります
Eventhouse への直接取り込み	connectionName と mappingRuleName が設定されている必要があります

EventstreamProperties の検証

eventstreamProperties.json をデコードして、次の項目を確認します。

• retentionTimeInDays が 1〜90 の範囲内であること
• eventThroughputLevel が Low、Medium、または High であること

---

注意点、ルール、トラブルシューティング

必ず行うこと

• az rest の呼び出しでは、必ず --resource https://api.fabric.microsoft.com を指定します
• ワークスペース名から ID、およびアイテム名から ID を解決する際は、必ず JMESPath フィルタリングを使用します
• トポロジを調べる前に、必ず定義ペイロードを Base64 デコードします（Topology API は JSON を直接返すため不要です）
• Custom Endpoint の接続詳細には、Topology API を使用します — POST .../getDefinition は空のプロパティを返します。GET .../topology を呼び出して sourceId を取得し、その後 GET .../sources/{sourceId}/connection を呼び出します
• 定義エンドポイントには POST を使用します — POST .../getDefinition（GET ではありません）、POST .../updateDefinition（PUT ではありません）。公式ドキュメントを参照してください。
• ページネーションに対応します — 一覧レスポンスに continuationUri があるか確認します
• LRO レスポンスをポーリングします — Get Definition が 202 Accepted を返す場合があります

推奨事項

• トポロジ JSON を構造化された出力にデコードし、読みやすい概要を作成します
• 解析には jq（bash）または ConvertFrom-Json（PowerShell）を使用します
• ユーザーに問題を報告する前に、構成を検証します
• 宛先を後続のスキル（eventhouse、sqldw、spark）と照合します

避けること

• Eventstream と Eventhouse を混同しないでください。これらは Fabric の別々のワークロードです
• ワークスペース ID やアイテム ID をハードコードしないでください。必ず API を通じて取得します
• すべてのソースの種類が API の列挙型に含まれると想定しないでください。プレビュー段階のソースは UI にのみ存在します
• この利用スキルで Eventstream のトポロジを変更しないでください。書き込みには eventstream-authoring-cli を使用します
• Eventstream API を通じてイベントデータをクエリしないでください。取り込み済みデータのクエリには、後続のスキル（eventhouse-consumption-cli、sqldw-consumption-cli）を使用します

---

例

プラットフォームに関する注意 — 例では PowerShell を使用します。JSON 本文は必ず [IO.File]::WriteAllText() を使って一時ファイルに書き込み（BOM なし）、インラインの --body "..." ではなく --body "@$file" を az rest に渡してください。インラインの --body "..." は cmd.exe によって変更される場合があります。改行の問題を避けるため、 ConvertTo-Json には -Compress を指定します。インラインで安全に指定できる唯一の例外は、 本文が空の場合の --body '{}' です。大規模なワークスペースでは、ページネーションに対応するため、 一覧レスポンスに continuationUri があるか確認してください。

例 1：ワークスペース内のすべての Eventstream を一覧表示する

プロンプト：「分析ワークスペース内のすべての Eventstream を、名前と ID を表示して一覧表示してください。」

# 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

例 2：Eventstream のトポロジを調べる

プロンプト：「SensorIngestion Eventstream のトポロジを表示してください。すべてのソース、オペレーター、宛先を確認したいです。」

# 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))"
}

例 3：保持期間とスループットの設定を確認する

プロンプト：「SensorIngestion Eventstream の保持期間とスループットの設定を教えてください。」

注意：保持期間とスループットの設定は、定義の eventstreamProperties.json 部分に保存されています（eventstream.json ではありません）。この部分が存在しない場合、 Eventstream はプラットフォームの既定値を使用します。

# 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 via Location header)
$token = (az account get-access-token --resource "https://api.fabric.microsoft.com" --query accessToken -o tsv)
$ps5 = @{}; if ($PSVersionTable.PSVersion.Major -lt 6) { $ps5.UseBasicParsing = $true }
$response = Invoke-WebRequest @ps5 -Method Post `
  -Uri "https://api.fabric.microsoft.com/v1/workspaces/$wsId/eventstreams/$esId/getDefinition" `
  -Headers @{ Authorization = "Bearer $token"; "Content-Type" = "application/json" } `
  -Body '{}'

if ($response.StatusCode -eq 202) {
    $location = $response.Headers['Location']
    if ($location -is [array]) { $location = $location[0] }
    if (-not $location) { throw "LRO response missing Location header" }
    $ra = $response.Headers['Retry-After']
    if ($ra -is [array]) { $ra = $ra[0] }
    $retryAfter = if ($ra) { [int]$ra } else { 5 }
    $def = $null
    for ($i = 0; $i -lt 12; $i++) {
        Start-Sleep -Seconds $retryAfter
        $poll = Invoke-RestMethod -Uri $location -Headers @{ Authorization = "Bearer $token" }
        if ($poll.status -eq 'Succeeded') {
            $def = Invoke-RestMethod -Uri "$location/result" `
              -Headers @{ Authorization = "Bearer $token" }
            break
        } elseif ($poll.status -in @('Failed', 'Cancelled')) {
            throw "getDefinition LRO $($poll.status): $($poll.error.message)"
        }
    }
    if (-not $def -or -not $def.definition) { throw "getDefinition LRO timed out (last status: $(if ($poll) { $poll.status } else { 'unknown' }))" }
} else {
    $def = $response.Content | ConvertFrom-Json
}

# 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)"
    }
}

例 4：Custom Endpoint の接続メタデータを取得する

プロンプト：「SensorIngestion Eventstream の Custom Endpoint ソースについて、Kafka 接続メタデータを取得してください。」

セキュリティ：接続エンドポイントはアクセスキーを返します。 呼び出す前にユーザーの確認を得て、生の認証情報をログに記録しないでください。

# 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"

例 5：Eventstream の構成を検証する

プロンプト：「SensorIngestion Eventstream に構成上の問題がないか確認してください。」

# 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"
}