Add AICL example: 59_streaming_pipeline.aicl

data/aicl/examples/59_streaming_pipeline.aicl

@@ -0,0 +1,402 @@
+# AICL Example: Streaming Data Pipeline
+# Kafka-like distributed streaming data pipeline with exactly-once semantics, schema registry,
+# stream-table duality, watermarking, and partitioned processing for event-driven architectures.
+
+Goal: Build a distributed streaming data pipeline that provides Kafka-like event streaming with
+    exactly-once delivery semantics, centralized schema management, stream-table duality for
+    stateful processing, and watermark-based time tracking for out-of-order event handling.
+
+Constraint: Exactly-once semantics must be guaranteed end-to-end from producer to consumer
+    through idempotent producers, transactional publishing, and consumer deduplication.
+
+Constraint: Schema registry must support backward, forward, and full compatibility modes
+    with zero-downtime schema evolution and automatic migration planning.
+
+Constraint: Stream-table duality must allow seamless switching between materialized table
+    views and changelog streams with point-in-time query capability.
+
+Constraint: Watermarking must track event time progress across all partitions with configurable
+    allowed lateness and idle partition detection.
+
+Risk: Producer sends duplicate messages during network retry causing duplicate processing
+Recovery: Enable idempotent producer with sequence number tracking per partition; consumer
+    deduplication using offset and message key tracking; compaction removes duplicates
+
+Risk: Schema evolution breaks deserialization for existing consumers
+Recovery: Schema registry rejects incompatible changes; consumers fall back to last compatible
+    schema version; dead-letter queue captures undecodable messages for manual resolution
+
+Risk: Partition rebalancing during consumer group changes causes processing gaps
+Recovery: Implement cooperative rebalancing with incremental partition assignment; consumers
+    complete in-flight processing before releasing partitions; offset commit on rebalance
+
+Risk: Watermark stalls on idle partitions blocking downstream window processing
+Recovery: Detect idle partitions via configurable timeout; advance watermark using active
+    partition progress; emit watermark advancement events to unblock downstream operators
+
+Risk: State store corruption during stream-table duality materialization
+Recovery: Maintain replicated state store with write-ahead log; recover from replica on
+    corruption; fallback to changelog topic replay for full state reconstruction
+
+Risk: Broker failure causes data unavailability for critical partitions
+Recovery: Replicate partitions across multiple brokers with ISR quorum; promote in-sync
+    replica to leader; under-replicated partition alert triggers automatic re-replication
+
+Layer: StreamingPipeline
+    SubLayer: EventBroker
+    SubLayer: SchemaRegistry
+    SubLayer: StreamProcessor
+    SubLayer: StateStore
+    SubLayer: WatermarkTracker
+
+Validation: Exactly-once delivery verified through end-to-end idempotency checks
+Validation: Schema compatibility must pass automated validation before registration
+Validation: Watermark must advance monotonically and never regress
+Validation: State store must be recoverable from changelog within 5 minutes
+Validation: Partition rebalancing must complete within 30 seconds without data loss
+Validation: Consumer lag must not exceed 1000 messages under normal operating conditions
+Validation: All produced messages must be readable by at least one in-sync replica
+
+# Level 2: Entities
+
+Entity Topic
+    topic_id: string
+    topic_name: string
+    partition_count: integer
+    replication_factor: integer
+    retention_ms: integer
+    cleanup_policy: string
+    schema_subject: string
+    is_compacted: boolean
+
+Entity Producer
+    producer_id: string
+    client_id: string
+    transactional_id: string
+    acks_mode: string
+    enable_idempotence: boolean
+    batch_size: integer
+    linger_ms: integer
+    schema_version: string
+
+Entity ConsumerGroup
+    group_id: string
+    members: list
+    partition_assignment: dict
+    commit_offset: dict
+    rebalance_strategy: string
+    session_timeout_ms: integer
+    is_rebalancing: boolean
+
+Entity SchemaVersion
+    schema_id: integer
+    subject: string
+    version: integer
+    schema_definition: dict
+    compatibility_mode: string
+    fingerprint: string
+    registered_at: datetime
+    is_latest: boolean
+
+Entity StateStore
+    store_id: string
+    store_name: string
+    store_type: string
+    changelog_topic: string
+    partition_count: integer
+    total_size_bytes: integer
+    last_checkpoint: datetime
+    is_persistent: boolean
+
+Entity Watermark
+    partition_id: integer
+    event_time: datetime
+    watermark_time: datetime
+    idle_detected: boolean
+    last_event_seen: datetime
+    allowed_lateness_ms: integer
+
+# Level 3: Behaviors
+
+Behavior PublishEvent
+    Input:
+        topic: string
+        key: string
+        value: bytes
+        headers: dict
+        event_timestamp: datetime
+    Output:
+        partition: integer
+        offset: integer
+        timestamp: datetime
+    Action:
+        Validate event against registered schema subject and version
+        Assign partition based on key hash or custom partitioner
+        Begin transaction if producer is transactional
+        Append to partition log with replication to ISR
+        Commit transaction and return partition-offset pair
+
+Behavior SubscribeToTopic
+    Input:
+        group_id: string
+        topics: list
+        rebalance_callback: string
+    Output:
+        assigned_partitions: list
+        committed_offsets: dict
+    Action:
+        Register consumer with group coordinator
+        Participate in cooperative rebalancing protocol
+        Receive partition assignments with committed offsets
+        Initialize fetch loops for assigned partitions
+        Report heartbeat and process records in poll loop
+
+Behavior RegisterSchema
+    Input:
+        subject: string
+        schema_definition: dict
+        compatibility_mode: string
+    Output:
+        schema_id: integer
+        version: integer
+        is_compatible: boolean
+    Action:
+        Compute schema fingerprint for deduplication check
+        Validate compatibility against existing versions per mode
+        Register new version if compatible, reject otherwise
+        Broadcast schema update to subscribed producers and consumers
+        Update schema cache on all broker nodes
+
+Behavior MaterializeTable
+    Input:
+        source_topic: string
+        state_store_name: string
+        key_extractor: string
+        value_aggregator: string
+    Output:
+        store_id: string
+        record_count: integer
+        size_bytes: integer
+    Action:
+        Subscribe to source topic changelog from beginning
+        Apply key extraction and value aggregation per record
+        Update state store with upsert or delete operations
+        Checkpoint state periodically to changelog topic
+        Expose point-in-time query interface over materialized state
+
+Behavior AdvanceWatermark
+    Input:
+        partition_id: integer
+        event_timestamp: datetime
+    Output:
+        new_watermark: datetime
+        idle_partitions: list
+    Action:
+        Track per-partition event time progress
+        Detect idle partitions exceeding idle timeout
+        Compute global watermark as min across active partitions
+        Emit watermark advancement events downstream
+        Handle late-arriving events within allowed lateness window
+
+# Level 4: Conditions
+
+Condition: SchemaIncompatible
+    When new schema version fails compatibility check against latest version
+    Then reject schema registration and return detailed incompatibility report
+        with specific fields causing the violation and suggested modifications
+
+Condition: ConsumerLagExceeded
+    When consumer group lag exceeds 10000 messages on any partition
+    Then scale up consumer instances within the group, increase fetch rate,
+        and alert operations team if lag continues to grow for 5 minutes
+
+Condition: UnderReplicatedPartition
+    When partition ISR count drops below configured replication factor
+    Then prioritize replica recovery for under-replicated partitions, alert
+        operations team, and restrict leader election to in-sync replicas only
+
+Condition: IdlePartitionDetected
+    When partition receives no events for longer than idle_timeout_ms
+    Then mark partition as idle, exclude from watermark computation, and
+        advance watermark using only active partitions to prevent stalling
+
+Condition: StateStoreCorrupted
+    When state store checksum validation fails on checkpoint load
+    Then initiate recovery from changelog topic replay, redirect queries
+        to replica store during recovery, and log corruption details for analysis
+
+# Level 5: Events
+
+Event: SchemaRegistered
+    On new_schema_version_approved
+    Action: Broadcast schema update to all registered producers and consumers,
+        update schema cache on brokers, and validate existing data compatibility
+
+Event: RebalanceStarted
+    On consumer_group_rebalance_triggered
+    Action: Pause record processing for partitions being revoked, commit current
+        offsets, execute cooperative rebalancing protocol, and resume with new assignments
+
+Event: WatermarkAdvanced
+    On global_watermark_progress_updated
+    Action: Trigger downstream window evaluations, emit late-arrival corrections
+        for events between old and new watermark, and update watermark metrics
+
+Event: StateCheckpointed
+    On state_store_checkpoint_completed
+    Action: Commit consumer offsets to match checkpoint, compact changelog topic
+        up to checkpoint offset, and log checkpoint metadata for recovery
+
+Event: TransactionCompleted
+    On producer_transaction_commit_or_abort
+    Action: Make transactional records visible to consumers, release partition
+        locks held by transaction, and update transaction coordinator state
+
+# Level 6: Concurrency
+
+Parallel:
+    Event production across all topic partitions
+    Consumer group record processing per partition
+    Schema validation and registration
+    State store materialization and checkpointing
+    Watermark tracking across all partitions
+    Partition replication and ISR management
+    Changelog compaction for state stores
+
+# Level 7: Optimization
+
+Optimize: Throughput and latency of event streaming pipeline
+    Priority: Producer batching optimized for throughput without exceeding linger latency
+    Priority: Consumer prefetch and parallel processing within partition assignment
+    Priority: State store checkpoint frequency balanced against recovery time
+    Priority: Schema cache locality on brokers closest to producers and consumers
+    Priority: Changelog compaction scheduled during low-traffic periods
+
+# Level 8: Learning
+
+Learn: Optimal partition count per topic based on throughput patterns
+    Goal: Minimize consumer group rebalancing while maximizing parallelism
+    Adapt: Partition count recommendations for new topics
+        Based: Historical throughput per partition, consumer group size, and rebalance frequency
+
+Learn: Consumer fetch parameters for optimal throughput
+    Goal: Achieve target throughput with minimum consumer CPU utilization
+    Adapt: Fetch size, max poll records, and processing concurrency
+        Based: Consumer processing latency, message size distribution, and lag patterns
+
+Learn: Watermark idle timeout configuration per topic
+    Goal: Advance watermarks promptly without excessive late-arrival corrections
+    Adapt: Idle partition timeout and allowed lateness parameters
+        Based: Event arrival rate distribution, late arrival frequency, and window accuracy
+
+# Level 9: Security
+
+Security:
+    Encrypt: All inter-broker communication using TLS with mutual authentication
+    Encrypt: Data at rest on broker log segments using AES-256 with per-topic keys
+    Protect: Schema registry access with RBAC and schema-level permissions
+    Protect: Consumer group operations with ACLs per principal and topic
+    Protect: State store contents with encryption and access control
+    Encrypt: Producer-consumer communication with SASL/OAUTHBEARER and TLS
+
+# Level 10: Native
+
+Native: Java
+{
+import org.apache.kafka.clients.producer.*;
+import org.apache.kafka.clients.consumer.*;
+import org.apache.kafka.common.*;
+import org.apache.kafka.streams.*;
+import org.apache.kafka.streams.processor.*;
+import org.apache.kafka.streams.state.*;
+import java.time.Duration;
+import java.util.*;
+
+public class StreamingPipeline {
+
+    private final KafkaProducer<String, byte[]> producer;
+    private final KafkaStreams streams;
+    private final SchemaRegistryClient schemaRegistry;
+
+    public StreamingPipeline(Properties producerConfig,
+                              Properties streamsConfig,
+                              SchemaRegistryClient schemaRegistry) {
+        this.producer = new KafkaProducer<>(producerConfig);
+        this.schemaRegistry = schemaRegistry;
+        this.streams = null;
+    }
+
+    public RecordMetadata publishEvent(String topic, String key, byte[] value,
+                                        Map<String, String> headers) throws Exception {
+        String subject = topic + "-value";
+        int schemaId = schemaRegistry.getLatestVersion(subject).getId();
+
+        ProducerRecord<String, byte[]> record = new ProducerRecord<>(topic, key, value);
+        headers.forEach(record.headers()::add);
+
+        try {
+            return producer.send(record).get();
+        } catch (Exception e) {
+            producer.abortTransaction();
+            throw new PipelineException("Failed to publish event", e);
+        }
+    }
+
+    public KafkaStreams buildMaterializedView(String sourceTopic,
+                                               String storeName) {
+        Properties props = new Properties();
+        props.put(StreamsConfig.APPLICATION_ID_CONFIG, storeName + "-app");
+        props.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "localhost:9092");
+
+        StreamsBuilder builder = new StreamsBuilder();
+        builder.table(sourceTopic,
+            Materialized.<String, byte[], KeyValueStore<Bytes, byte[]>>as(storeName)
+                .withKeySerde(Serdes.String())
+                .withValueSerde(Serdes.ByteArray())
+                .withCachingEnabled()
+                .withLoggingEnabled(Collections.emptyMap())
+        );
+
+        KafkaStreams streams = new KafkaStreams(builder.build(), props);
+        streams.setStateListener((newState, oldState) -> {
+            if (newState == KafkaStreams.State.RUNNING) {
+                System.out.println("State store " + storeName + " materialized");
+            }
+        });
+        return streams;
+    }
+
+    public static class WatermarkTracker {
+        private final Map<TopicPartition, Long> partitionWatermarks = new HashMap<>();
+        private final Duration allowedLateness;
+        private final Duration idleTimeout;
+        private final Map<TopicPartition, Long> lastEventTime = new HashMap<>();
+
+        public WatermarkTracker(Duration allowedLateness, Duration idleTimeout) {
+            this.allowedLateness = allowedLateness;
+            this.idleTimeout = idleTimeout;
+        }
+
+        public synchronized long advanceWatermark(TopicPartition tp, long eventTime) {
+            lastEventTime.put(tp, System.currentTimeMillis());
+            partitionWatermarks.merge(tp, eventTime, Math::max);
+            return computeGlobalWatermark();
+        }
+
+        private long computeGlobalWatermark() {
+            long now = System.currentTimeMillis();
+            long globalWatermark = Long.MAX_VALUE;
+            for (Map.Entry<TopicPartition, Long> entry : partitionWatermarks.entrySet()) {
+                long lastSeen = lastEventTime.getOrDefault(entry.getKey(), now);
+                if (now - lastSeen > idleTimeout.toMillis()) {
+                    continue;
+                }
+                globalWatermark = Math.min(globalWatermark, entry.getValue());
+            }
+            return globalWatermark == Long.MAX_VALUE
+                ? partitionWatermarks.values().stream().max(Long::compare).orElse(0L)
+                : globalWatermark;
+        }
+    }
+}
+}