Azure Hdinsight Migration Kafka To Fabric Rti

Migrating HDInsight Kafka → Microsoft Fabric Real-Time Intelligence

A field guide for migrating production Apache Kafka workloads from Azure HDInsight Kafka clusters to Microsoft Fabric Real-Time Intelligence (RTI) — Eventstreams, Eventhouse / KQL Database, Activator, and Real-Time Dashboards — with Azure Event Hubs (Kafka endpoint) as the recommended message-bus replacement when downstream producers/consumers must keep speaking Kafka protocol.

Why this migration? HDInsight Kafka 4.0 reached end-of-standard support; 5.x is the last classic generation. Microsoft's recommended landing zones are:

• Azure Event Hubs (Kafka endpoint) — drop-in for the broker tier (keeps Kafka clients).
• Fabric Eventstream — managed ingestion + routing + transformation (no broker management).
• Fabric Eventhouse / KQL DB — analytics + storage tier (replaces Spark/Hive on stream).
• Fabric Activator — alerting / triggers (replaces custom Storm/Spark Streaming alerters).

1. Decision Matrix — Which Fabric Path?

Decision shortcut: If the workload's downstream is analytics + dashboards in the Fabric/Power BI ecosystem, go full Eventstream → Eventhouse. If it's an event bus serving many polyglot clients, go Event Hubs Kafka endpoint and bridge selected topics into Fabric Eventstream for analytics.

2. Assessment Phase

2.1 Inventory the HDInsight Kafka Cluster

Run on a head/broker node (SSH).

ssh sshuser@<cluster>-ssh.azurehdinsight.net

# Cluster + broker metadata
curl -s -u admin:$PASS \
  "https://<cluster>.azurehdinsight.net/api/v1/clusters/<cluster>" \
  | python3 -m json.tool > cluster_meta.json

# Broker list (resolve once, used everywhere)
BROKERS=$(curl -s -u admin:$PASS \
  "https://<cluster>.azurehdinsight.net/api/v1/clusters/<cluster>/services/KAFKA/components/KAFKA_BROKER" \
  | python3 -c "import sys,json; d=json.load(sys.stdin); print(','.join([h['HostRoles']['host_name']+':9092' for h in d['host_components']]))")
echo $BROKERS > brokers.txt

KBIN=/usr/hdp/current/kafka-broker/bin

# Kafka version
$KBIN/kafka-topics.sh --version | tee kafka_version.txt

# All topics + per-topic config
$KBIN/kafka-topics.sh --bootstrap-server $BROKERS --list > topics.txt
while read t; do
  $KBIN/kafka-topics.sh --bootstrap-server $BROKERS --describe --topic "$t" \
    >> topic_describe.txt
done < topics.txt

# Topic-level configs (retention, compaction, etc.)
$KBIN/kafka-configs.sh --bootstrap-server $BROKERS --entity-type topics --describe \
  > topic_configs.txt

# Broker-level configs
$KBIN/kafka-configs.sh --bootstrap-server $BROKERS --entity-type brokers --describe \
  > broker_configs.txt

# Consumer groups + lag
$KBIN/kafka-consumer-groups.sh --bootstrap-server $BROKERS --list > consumer_groups.txt
while read g; do
  echo "=== $g ===" >> consumer_groups_describe.txt
  $KBIN/kafka-consumer-groups.sh --bootstrap-server $BROKERS \
    --describe --group "$g" >> consumer_groups_describe.txt
done < consumer_groups.txt

# ACLs (if SASL/SSL enabled — ESP clusters)
$KBIN/kafka-acls.sh --bootstrap-server $BROKERS --list > acls.txt 2>/dev/null || true

# Per-topic size + message count estimate
$KBIN/kafka-log-dirs.sh --bootstrap-server $BROKERS --describe \
  | python3 -c "
import sys, json, re
data = ''.join(l for l in sys.stdin if l.startswith('{'))
d = json.loads(data)
totals = {}
for b in d['brokers']:
  for ld in b['logDirs']:
    for p in ld['partitions']:
      topic = p['partition'].rsplit('-', 1)[0]
      totals[topic] = totals.get(topic, 0) + p['size']
for t, s in sorted(totals.items(), key=lambda x: -x[1]):
  print(f'{s/1e9:8.2f} GB  {t}')
" > topic_sizes.txt

2.2 Inventory the Producer / Consumer Codebase

cd /path/to/repos

# 1) Bootstrap server / connection strings
grep -RInE 'bootstrap[_.-]?servers|kafka\.host|broker[_.-]?list' \
  --include='*.py' --include='*.java' --include='*.scala' --include='*.go' \
  --include='*.js' --include='*.ts' --include='*.cs' --include='*.yaml' --include='*.yml'

# 2) Auth mechanisms in use
grep -RInE 'security\.protocol|sasl\.mechanism|sasl\.jaas\.config|kerberos|GSSAPI|PLAIN|SCRAM' \
  --include='*.properties' --include='*.conf' --include='*.py' --include='*.java'

# 3) Schema Registry usage (Confluent / Apicurio)
grep -RInE 'schema[_.-]?registry|confluent[_.-]?serializer|KafkaAvroSerializer|apicurio' \
  --include='*.py' --include='*.java' --include='*.scala' --include='*.properties'

# 4) Kafka Streams / ksqlDB
grep -RInE 'org\.apache\.kafka\.streams|StreamsBuilder|ksql' \
  --include='*.java' --include='*.scala' --include='*.sql'

# 5) Kafka Connect connectors
find . -name 'connect-*.properties' -o -name 'connector*.json' \
  -o -name 'sink-*.json' -o -name 'source-*.json'

# 6) MirrorMaker configs
find . -name 'mm2.properties' -o -name 'mirror-maker*.properties' \
  -o -name 'connect-mirror-source*.properties'

# 7) Spark Streaming consumers (HDI Spark talking to HDI Kafka)
grep -RInE 'spark\.readStream.*kafka|KafkaUtils\.create' \
  --include='*.py' --include='*.scala'

# 8) Storm topologies
find . -name 'topology*.yaml' -o -name '*.flux'

2.3 Categorize Each Workload

Build a CSV: topic, throughput_mbps, retention_h, compacted, partitions, replication, producers, consumers, schema_type, target, complexity.

3. Component Mapping

3.1 Brokers / Topics

3.2 Streaming Compute

3.3 Storage / Sinks

4. Topic + Schema Migration

4.1 Topic-by-Topic Provisioning Script

Create equivalent Event Hubs from your topic inventory.

# Pre-reqs: Event Hubs namespace exists, az CLI logged in
NS=my-eh-namespace
RG=eh-rg

# Read topics.txt + topic_describe.txt and create event hubs
while read topic; do
  # Pull partition count from describe output
  partitions=$(grep -A1 "Topic: $topic\b" topic_describe.txt | head -1 | \
               grep -oE 'PartitionCount: [0-9]+' | awk '{print $2}')
  retention_h=$(grep -B0 -A0 "^Topic: $topic\b" topic_configs.txt | \
               grep -oE 'retention.ms=[0-9]+' | head -1 | cut -d= -f2)
  retention_d=$(( ${retention_h:-86400000} / 86400000 ))
  retention_d=$(( retention_d < 1 ? 1 : retention_d ))

  echo "Creating EH '$topic' with $partitions partitions, $retention_d days retention"

  az eventhubs eventhub create \
    --resource-group $RG \
    --namespace-name $NS \
    --name "$topic" \
    --partition-count "${partitions:-4}" \
    --retention-time "$retention_d" \
    --cleanup-policy Delete \
    --status Active \
    --output none
done < topics.txt

Naming gotcha: Event Hub names must match ^[a-zA-Z0-9][\w.-]{0,255}$. Kafka topic names with _ / . mix are usually fine, but topics starting with _ (Confluent internal, __consumer_offsets) cannot be created — you don't need to migrate those.

4.2 Schema Migration

Recommended: For new ingestion paths, prefer JSON Lines or Avro single-object encoded (schema embedded). This eliminates the Schema Registry dependency that Fabric has no managed equivalent for.

4.3 Configure Producers/Consumers for Event Hubs Kafka Endpoint

# OLD — HDInsight Kafka (PLAINTEXT or SASL/GSSAPI)
bootstrap.servers=broker0:9092,broker1:9092,broker2:9092
security.protocol=PLAINTEXT

# NEW — Event Hubs Kafka endpoint (SASL_SSL + connection string)
bootstrap.servers=my-eh-namespace.servicebus.windows.net:9093
security.protocol=SASL_SSL
sasl.mechanism=PLAIN
sasl.jaas.config=org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="$ConnectionString" \
  password="Endpoint=sb://my-eh-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=...";

# Also recommended:
session.timeout.ms=30000
request.timeout.ms=60000   # EH default broker-side is high; client should match

# Python (confluent-kafka or kafka-python) — Event Hubs Kafka endpoint
from confluent_kafka import Producer
producer = Producer({
    'bootstrap.servers': 'my-eh-namespace.servicebus.windows.net:9093',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism':    'PLAIN',
    'sasl.username':     '$ConnectionString',
    'sasl.password':     'Endpoint=sb://my-eh-namespace...;SharedAccessKey=...',
})
producer.produce('orders', key='42', value=b'{"order_id":42}')
producer.flush()

# Better: Entra ID OAuth (no static keys; aligns with Managed Identity story)
from confluent_kafka import Producer
from azure.identity import DefaultAzureCredential

cred = DefaultAzureCredential()

def oauth_cb(_):
    token = cred.get_token("https://eventhubs.azure.net/.default")
    return token.token, token.expires_on

producer = Producer({
    'bootstrap.servers': 'my-eh-namespace.servicebus.windows.net:9093',
    'security.protocol': 'SASL_SSL',
    'sasl.mechanism':    'OAUTHBEARER',
    'oauth_cb':          oauth_cb,
})

Gotcha: Event Hubs Kafka endpoint is TLS-only (port 9093, never 9092). Plain PLAINTEXT configurations from HDI must be changed.

5. Cutover Strategy: MirrorMaker 2 Bridge

The safest cutover preserves both clusters in parallel using MirrorMaker 2 (MM2) to replicate from HDI Kafka → Event Hubs Kafka endpoint, letting you cut producers and consumers over independently.

5.1 MM2 Source-Target Config

Run MM2 from a small AKS pod / VM that has network access to both endpoints.

# mm2.properties
clusters = hdi, eh

hdi.bootstrap.servers = broker0:9092,broker1:9092,broker2:9092
hdi.security.protocol = PLAINTEXT
# (or SASL/GSSAPI if HDI is ESP)

eh.bootstrap.servers  = my-eh-namespace.servicebus.windows.net:9093
eh.security.protocol  = SASL_SSL
eh.sasl.mechanism     = PLAIN
eh.sasl.jaas.config   = org.apache.kafka.common.security.plain.PlainLoginModule required \
  username="$ConnectionString" \
  password="Endpoint=sb://my-eh-namespace.servicebus.windows.net/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=...";

# Direction: hdi → eh (one-way)
hdi->eh.enabled = true
hdi->eh.topics = .*
hdi->eh.topics.exclude = __consumer_offsets,_schemas,connect-.*

# Replication factor on target — EH manages internally, but client config still needs it
replication.factor = 3
checkpoints.topic.replication.factor = 3
heartbeats.topic.replication.factor  = 3
offset-syncs.topic.replication.factor= 3

# Translate consumer group offsets too
hdi->eh.sync.group.offsets.enabled = true
hdi->eh.emit.checkpoints.enabled   = true

# Don't auto-rename topics with cluster prefix (default would be "hdi.orders")
hdi->eh.replication.policy.class = org.apache.kafka.connect.mirror.IdentityReplicationPolicy

# Start MM2 (Kafka 2.7+ ships connect-mirror-maker.sh)
$KBIN/connect-mirror-maker.sh mm2.properties

Critical gotcha: The default DefaultReplicationPolicy prefixes topics with the source cluster alias (e.g., hdi.orders). For an in-place migration where consumers point at the same topic name, you must set IdentityReplicationPolicy. Test consumer behavior carefully — IdentityReplicationPolicy is incompatible with bidirectional replication.

5.2 Cutover Sequence

T-7d  : Stand up Event Hubs namespace + provision all topics
T-6d  : Start MM2 hdi → eh; verify lag stays ~0
T-5d  : Cut consumers ONE GROUP AT A TIME to EH endpoint
        - Stop HDI consumer
        - Use MM2 checkpoint to translate offset (see 5.3)
        - Start EH consumer at translated offset
        - Validate counts, then process forward
T-2d  : All consumers on EH; HDI consumer-side empty
T-1d  : Cut PRODUCERS to EH endpoint (config swap + restart)
        - Stop HDI producer
        - Wait for MM2 to drain remaining records
        - Start EH producer
T-0   : Stop MM2; HDI cluster has no traffic
T+7d  : Tear down HDI Kafka cluster (keep ADLS / commit offsets backup)

5.3 Translating Consumer-Group Offsets

MM2 maintains an offset-syncs internal topic that maps source offsets → target offsets. On cutover, query the checkpoint to get the target offset for each consumer group:

$KBIN/kafka-console-consumer.sh \
  --bootstrap-server $EH_BROKERS --consumer.config eh-client.properties \
  --topic mm2-checkpoints.hdi.internal \
  --from-beginning --max-messages 1000 \
  --formatter org.apache.kafka.connect.mirror.formatters.CheckpointFormatter \
  > checkpoints.txt

# Each line: group=X, topic=Y, partition=Z, upstream-offset=A, downstream-offset=B
# Use B (downstream-offset) as the seek target on the EH consumer

For each consumer group, after stopping it on HDI:

from confluent_kafka import Consumer, TopicPartition
c = Consumer({
    'bootstrap.servers': 'my-eh-namespace.servicebus.windows.net:9093',
    'security.protocol': 'SASL_SSL', 'sasl.mechanism': 'PLAIN',
    'sasl.username': '$ConnectionString', 'sasl.password': '...',
    'group.id': 'orders-consumer-v1',
    'enable.auto.commit': False,
})

# Set group offsets from MM2 checkpoint
checkpoints = {('orders', 0): 1024567, ('orders', 1): 1019988, ...}
tps = [TopicPartition(t, p, off) for (t, p), off in checkpoints.items()]
c.commit(offsets=tps, asynchronous=False)
print("Offsets committed; safe to start consumer normally")

5.4 Producer Cutover Idempotence

Use producer transactions OR idempotent producers during cutover to avoid duplicates if a producer has in-flight records during config swap:

enable.idempotence=true
acks=all
max.in.flight.requests.per.connection=5
retries=2147483647

6. Fabric Eventstream Wiring

6.1 Create Eventstream Pointing at Event Hubs

# Fabric REST API — create Eventstream with EH source
import requests
ws = "<workspace-id>"
hdrs = {"Authorization": f"Bearer {fabric_token}", "Content-Type": "application/json"}

resp = requests.post(
    f"https://api.fabric.microsoft.com/v1/workspaces/{ws}/eventstreams",
    headers=hdrs,
    json={
        "displayName": "orders_stream",
        "description": "Orders ingestion from EH",
    },
)
es_id = resp.json()["id"]

# Add EH source via topology PATCH (UI-driven flow shown for reference)
# Topology JSON includes:
#   sources:        [{name: "orders_eh", type: "AzureEventHub",
#                     properties: {endpoint, eventHubName, sharedAccessKeyName,
#                                  sharedAccessKey OR managedIdentity}}]
#   destinations:   [{name: "to_eventhouse", type: "Eventhouse",
#                     properties: {workspaceId, eventhouseId, kqlDatabase, table}}]
#   operators:      [{name: "filter1", type: "Filter", inputs:["orders_eh"],
#                     properties: {expression: "amount > 0"}}, ...]
#   streams:        edges connecting sources → operators → destinations

6.2 Eventstream Sources (Replacing HDI Kafka Producers)

6.3 Eventstream Destinations (Replacing HDI Spark Sinks)

6.4 Eventstream Event Processor (Replacing ksqlDB / Light Spark Logic)

Drag-drop operators replace simple stream transformations:

Limitation: Eventstream Event Processor handles single-event and windowed aggregations well, but stream-stream joins and stateful exactly-once logic belong in Spark Structured Streaming or KQL update policies.

7. Eventhouse / KQL Database Setup

7.1 Create Eventhouse + Database

# Fabric REST — create eventhouse (bundle of KQL DB + queryset)
resp = requests.post(
    f"https://api.fabric.microsoft.com/v1/workspaces/{ws}/eventhouses",
    headers=hdrs,
    json={"displayName": "orders_eh", "description": "Orders telemetry"},
)

7.2 KQL Table + Ingestion Mapping

// Create table matching expected ingest schema
.create table OrdersRaw (
    EventTime: datetime,
    OrderId:   long,
    CustomerId:long,
    Amount:    real,
    Status:    string,
    Region:    string,
    RawJson:   dynamic
)

// JSON ingestion mapping — maps Eventstream JSON → table columns
.create table OrdersRaw ingestion json mapping 'OrdersRawMapping'
'['
'  {"column":"EventTime",  "Properties":{"Path":"$.event_time","Transform":"DateTimeFromUnixSeconds"}},'
'  {"column":"OrderId",    "Properties":{"Path":"$.order_id"}},'
'  {"column":"CustomerId", "Properties":{"Path":"$.customer_id"}},'
'  {"column":"Amount",     "Properties":{"Path":"$.amount"}},'
'  {"column":"Status",     "Properties":{"Path":"$.status"}},'
'  {"column":"Region",     "Properties":{"Path":"$.region"}},'
'  {"column":"RawJson",    "Properties":{"Path":"$"}}'
']'

// Streaming ingestion policy — required for sub-5s latency from Eventstream
.alter table OrdersRaw policy streamingingestion enable

7.3 Update Policies (Replacing ksqlDB / Spark Stream Transformations)

KQL update policies run automatically on every ingestion, populating downstream "silver/gold" tables — equivalent to a Spark Structured Streaming aggregation but declarative and serverless.

// Silver: cleansed
.create table OrdersSilver (
    EventTime: datetime, OrderId: long, CustomerId: long,
    Amount: decimal, Status: string, Region: string
)

.create-or-alter function ParseRawOrders() {
    OrdersRaw
    | where isnotempty(OrderId) and Amount > 0
    | extend Amount = todecimal(Amount)
    | project EventTime, OrderId, CustomerId, Amount, Status, Region
}

.alter table OrdersSilver policy update
'[{"Source": "OrdersRaw", "Query": "ParseRawOrders()", "IsEnabled": true, "IsTransactional": true}]'

// Gold: aggregated by 5-minute tumbling window — runs automatically
.create table OrdersAgg5m (
    Window: datetime, Region: string, OrderCount: long, Revenue: decimal
)

.create-or-alter function AggOrders5m() {
    OrdersSilver
    | summarize OrderCount = count(), Revenue = sum(Amount)
        by Window = bin(EventTime, 5m), Region
}

// Drive aggregation via materialized view (preferred over update policy for aggregations)
.create materialized-view AggOrders5mMV on table OrdersSilver
{
    OrdersSilver
    | summarize Revenue = sum(Amount), OrderCount = count()
        by Window = bin(EventTime, 5m), Region
}

7.4 Spark Structured Streaming Equivalent (When KQL Isn't Enough)

Some HDI Spark patterns (stateful joins, complex windowing, ML scoring inline) need to stay in Spark. Run them in a Fabric Spark notebook consuming from Event Hubs:

# Fabric notebook — Spark Structured Streaming from Event Hubs Kafka endpoint
EH_BOOTSTRAP = "my-eh-namespace.servicebus.windows.net:9093"
EH_CONNSTR   = mssparkutils.credentials.getSecret("https://my-kv.vault.azure.net/", "eh-conn")
JAAS = (f'org.apache.kafka.common.security.plain.PlainLoginModule required '
        f'username="$ConnectionString" password="{EH_CONNSTR}";')

stream = (spark.readStream
    .format("kafka")
    .option("kafka.bootstrap.servers", EH_BOOTSTRAP)
    .option("subscribe", "orders")
    .option("kafka.security.protocol", "SASL_SSL")
    .option("kafka.sasl.mechanism", "PLAIN")
    .option("kafka.sasl.jaas.config", JAAS)
    .option("startingOffsets", "earliest")
    .load()
    .selectExpr("CAST(value AS STRING) as json", "timestamp")
)

from pyspark.sql.functions import from_json
from pyspark.sql.types import *
schema = StructType([
    StructField("order_id",    LongType()),
    StructField("customer_id", LongType()),
    StructField("amount",      DoubleType()),
    StructField("status",      StringType()),
    StructField("region",      StringType()),
])

parsed = stream.select(
    from_json("json", schema).alias("d"), "timestamp"
).select("d.*", "timestamp")

# Write to Lakehouse Delta with availableNow trigger (cost-efficient)
(parsed.writeStream
    .format("delta")
    .outputMode("append")
    .option("checkpointLocation", "Files/_checkpoints/orders_bronze/")
    .trigger(availableNow=True)
    .toTable("sales_lakehouse.orders_bronze"))

8. Security Mapping

8.1 Authentication

Action item: Remove kinit, krb5.conf, JAAS Kerberos sections, and keytab files from your producer/consumer hosts. Replace with either SAS connection strings (simple) or OAuth via DefaultAzureCredential (preferred — no rotation).

8.2 Authorization

# Grant a managed identity Send rights on a specific event hub
az role assignment create \
  --assignee $MI_OBJECT_ID \
  --role "Azure Event Hubs Data Sender" \
  --scope "/subscriptions/$SUB/resourceGroups/$RG/providers/Microsoft.EventHub/namespaces/$NS/eventhubs/orders"

8.3 Network

9. Real-Time Dashboards & Alerting

9.1 Replacing HDI Spark → Power BI Streaming Datasets

Old pattern: Spark Structured Streaming pushes to Power BI streaming dataset via REST.

New pattern (faster, no code):

Eventstream → KQL DB → Real-Time Dashboard (sub-second refresh)
                    OR Power BI Direct Query / Direct Lake on KQL DB

// KQL query powering a Real-Time Dashboard tile
OrdersSilver
| where EventTime > ago(1h)
| summarize Revenue = sum(Amount) by bin(EventTime, 1m), Region
| render timechart

9.2 Replacing Custom Storm/Spark Alerters with Activator

Eventstream → Activator (no-code rule):
  WHEN Amount > 10000 OR Status == 'fraud_suspected'
  WITHIN 1 minute, GROUPED BY CustomerId
  THEN POST to https://hooks.example.com/fraud
       AND email security@example.com
       AND post Teams channel #alerts

Activator replaces hand-rolled Spark Streaming alerter pipelines, with built-in de-duplication, rate-limiting, and audit trail.

10. Validation

10.1 Per-Topic Reconciliation

Run for each topic during MM2 bridge phase:

# Source: HDI Kafka — count records over a fixed window
SOURCE_COUNT=$($KBIN/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --bootstrap-server $HDI_BROKERS --topic orders --time -1 \
  | awk -F: '{sum += $3} END {print sum}')

# Target: Event Hubs — same approach
TARGET_COUNT=$($KBIN/kafka-run-class.sh kafka.tools.GetOffsetShell \
  --bootstrap-server $EH_BROKERS --command-config eh-client.properties \
  --topic orders --time -1 \
  | awk -F: '{sum += $3} END {print sum}')

echo "Source: $SOURCE_COUNT / Target: $TARGET_COUNT  diff: $((SOURCE_COUNT - TARGET_COUNT))"

10.2 Content Equality (Sample-Based)

# Pull last 1000 records from each side, compare by (key, value-hash)
# (use a small kafka client — confluent-kafka or kafka-python)
def fetch_sample(brokers, props, topic, n=1000):
    from confluent_kafka import Consumer, TopicPartition
    c = Consumer({**props, 'bootstrap.servers': brokers, 'group.id': f'verify-{topic}',
                  'auto.offset.reset': 'latest', 'enable.auto.commit': False})
    parts = c.list_topics(topic).topics[topic].partitions
    seek = []
    for pid in parts:
        tp = TopicPartition(topic, pid)
        lo, hi = c.get_watermark_offsets(tp)
        seek.append(TopicPartition(topic, pid, max(lo, hi - n // len(parts))))
    c.assign(seek)
    out = {}
    for _ in range(n):
        m = c.poll(2.0)
        if m is None: break
        out[(m.key(), hash(m.value()))] = (m.partition(), m.offset())
    c.close()
    return out

src = fetch_sample(HDI_BROKERS, {'security.protocol':'PLAINTEXT'}, 'orders')
tgt = fetch_sample(EH_BROKERS, EH_PROPS, 'orders')
intersection = set(src) & set(tgt)
print(f"src={len(src)} tgt={len(tgt)} intersection={len(intersection)} "
      f"agreement={len(intersection)/min(len(src),len(tgt)):.2%}")

10.3 Acceptance Criteria

migration_pass_criteria:
  data_correctness:
    - per_topic_count_diff_pct: < 0.01    # MM2 lag-tolerant
    - sample_content_agreement_pct: > 99.9
    - consumer_group_offset_translated: all
  performance:
    - producer_p99_latency_ms_delta_pct: < 30
    - consumer_p99_lag_seconds: < 10
  reliability:
    - mm2_bridge_uptime_7d_pct: > 99.9
    - eh_throttle_events_7d: 0
    - eventstream_run_state: Active
  security:
    - all_clients_use_tls: true
    - no_plaintext_endpoints: true
    - all_kerberos_keytabs_removed: true

11. Cost Re-Modeling

Sizing rule of thumb:

• HDI Kafka cluster of 3× D12 brokers (~24 TU equivalent, 30 MB/s ingress) ≈ EH Standard 20 TUs OR EH Premium PU=1 (better for >50 MB/s sustained).
• Eventstream throughput is bounded by Fabric capacity CU: ~50 MB/s per F64 capacity for typical JSON. For higher throughput, use EH directly + Fabric Spark consumer.

12. Pitfalls & Lessons Learned

1. Plain TCP / port 9092 will not work. Event Hubs Kafka endpoint is 9093 + TLS only. All client configs must change bootstrap.servers AND add SASL_SSL.

2. Event Hubs has no log compaction. If you rely on compacted topics (e.g., Kafka Streams state stores, change-log topics), you cannot drop in EH. Keep those topics on Kafka (managed Confluent on AKS) OR re-architect as KQL materialized-view / Lakehouse Delta MERGE.

3. __consumer_offsets doesn't migrate via MM2. MM2 mirrors checkpoints, not the __consumer_offsets topic itself. Use the offset translation pattern (§5.3) at cutover time.

4. Confluent Schema Registry has no managed equivalent. Either: keep SR on AKS, embed schema in payload (Avro single-object encoding), or move to schema-on-read in KQL with parse_json().

5. IdentityReplicationPolicy vs DefaultReplicationPolicy. Default prefixes topics with cluster alias (hdi.orders); Identity does not. Pick BEFORE starting MM2 — switching mid-flight orphans target offsets.

6. EH partition count is immutable on Basic and Standard tiers. Pick wisely. Premium and Dedicated tiers allow partition INCREASE (not decrease) after creation. Spark/Kafka consumer parallelism per group is bounded by partition count.

7. Idempotent producers + EH transactions. transactional.id is supported on EH (since 2023) but with caveats — EOS across topic boundaries may not match Kafka guarantees exactly. Test transaction-heavy producers thoroughly.

8. Kafka Streams apps don't run on Fabric. There is no "Fabric Streams" runtime. Three options: (a) keep Kafka Streams app on AKS pointed at EH; (b) rewrite as Fabric Spark Structured Streaming; (c) re-implement in KQL update policies for simple cases.

9. HDI ESP → no Kerberos support on EH. ESP-secured Kafka clients with GSSAPI must be re-configured with OAUTHBEARER (recommended) or PLAIN (with SAS). Remove krb5.conf, JAAS Kerberos sections, and keytab volumes.

10. MirrorMaker offset-syncs topic ACL. The MM2 process needs Send rights on mm2-offset-syncs.eh.internal, mm2-checkpoints.hdi.internal, and heartbeats topics on both sides. Easy to miss in IAM setup.

11. Eventstream operator language is not SQL. The drag-drop operators map to a proprietary expression language. Complex transformations belong in KQL update policies (declarative SQL-like) or Spark notebooks (PySpark).

12. Activator rules are not Spark UDFs. Activator handles per-event and windowed rule evaluation but cannot run arbitrary code. For ML-based alerting, score in Spark/MLflow and write the score back to KQL, then alert on the score column.

13. Geo-DR ≠ active-active. EH Geo-DR uses alias + paired secondary for failover; only one side is writable at a time. If you used MM2 for active-active multi-region with HDI, you must redesign — keep MM2 between two EH namespaces, or use Confluent Cluster Linking between Confluent Cloud regions.

14. JSON ingestion mapping is ordered. KQL JSON ingestion mappings match by Path, but column types are inferred at table-create time. Schema drift in incoming JSON silently drops/coerces fields. Snapshot a .show ingestion failures report daily during early operation.

15. KQL streaming ingestion has a ~5s latency floor. If you need <1s end-to-end (e.g., real-time fraud blocking), use a Spark Structured Streaming consumer that writes directly to a downstream system, not Eventstream → KQL.

16. Event Hubs send-batching matters. Default Kafka producer batches by size + linger.ms. For low-throughput topics, set linger.ms=20 to avoid sending one record per request — EH bills per-event AND per-operation.

17. MM2 throughput needs sizing. Each MM2 task handles N partitions. For a high-throughput topic, set tasks.max proportional to partitions. Default 1 is almost always too low — verify lag during initial sync.

13. Quick-Reference Migration Checklist

[ ] Inventory HDI Kafka: topics, configs, ACLs, consumer groups, sizes
[ ] Inventory clients: producer / consumer / Streams app code, schemas, auth modes
[ ] Categorize each topic: target = EH-only, Eventstream-bridge, or stay-on-Kafka
[ ] Decide schema strategy: keep SR on AKS, embed schema, or schema-on-read in KQL
[ ] Provision Event Hubs namespace (Standard or Premium based on throughput)
[ ] Provision per-topic event hubs (script from inventory)
[ ] Configure Entra ID + RBAC roles for producers/consumers (or generate SAS)
[ ] Stand up MM2 bridge HDI → EH, verify lag stays ~0 for ≥48h
[ ] Provision Fabric Eventhouse + KQL DB tables + ingestion mappings
[ ] Create Eventstream(s): EH source → operators → Eventhouse / Lakehouse destination
[ ] Validate per-topic counts + sample content equality
[ ] Cut consumers ONE GROUP AT A TIME, translate offsets, verify
[ ] Cut producers (config swap + restart with idempotent + acks=all)
[ ] Stop MM2; verify HDI cluster traffic dropped to zero
[ ] Switch dashboards: Spark/PBI streaming datasets → Real-Time Dashboards on KQL DB
[ ] Convert custom alerters → Activator rules
[ ] Schedule Eventhouse maintenance: retention policies, cache policy review
[ ] Set up alerts: EH throttling, Eventstream run-state, KQL ingestion failures
[ ] Tear down HDI Kafka cluster (keep ADLS / commit-offset backup for 30d)
[ ] Document new runbook + on-call procedures

14. Useful Commands & Snippets

# Get EH connection string (RootManageSharedAccessKey)
az eventhubs namespace authorization-rule keys list \
  --resource-group $RG --namespace-name $NS \
  --name RootManageSharedAccessKey \
  --query primaryConnectionString -o tsv

# List event hubs in a namespace
az eventhubs eventhub list -g $RG --namespace-name $NS --output table

# Tail an EH topic with kafkacat (kcat)
kcat -b my-eh-namespace.servicebus.windows.net:9093 \
     -X security.protocol=SASL_SSL \
     -X sasl.mechanisms=PLAIN \
     -X sasl.username='$ConnectionString' \
     -X sasl.password="$EH_CONNSTR" \
     -t orders -C -o end -q -J

# Check MM2 lag (it runs as a Connect cluster — query its REST API on port 8083)
curl -s http://mm2-host:8083/connectors/MirrorSourceConnector/status | python3 -m json.tool
curl -s http://mm2-host:8083/connectors/MirrorSourceConnector/topics

# KQL: ingestion failures (run in KQL queryset)
.show ingestion failures
| where FailedOn > ago(1h)
| summarize count() by Database, Table, Reason

Rules

1. **Never expose plai

…(truncated)