// Reusable pattern: Event-driven processing with orchestration
var tradingWorkflow = Temporal.WorkflowBuilder
    .OnKafkaEvent("trades")
    .FlinkProcess(env => env
        .FromKafka("raw-trades")  
        .Map(trade => trade.EnrichWithMarketData())
        .Rebalance()  // Dynamic scaling
        .Filter(trade => trade.PassesRiskChecks())
        .ToKafka("validated-trades"))
    .OrchestrateLongRunning(async () => {
        await settleTradeAsync();
        await updatePortfolioAsync(); 
        await generateRegulatoryReportAsync();
    });

// Reusable pattern: Multi-channel aggregation with coordination
var orderWorkflow = Temporal.WorkflowBuilder
    .OnMultipleKafkaEvents("web-orders", "mobile-orders", "pos-orders")
    .FlinkProcess(env => env
        .UnionStreams("web-orders", "mobile-orders", "pos-orders")
        .Map(order => order.Normalize())
        .KeyBy(order => order.CustomerId)
        .Window(TimeWindow.Of(Time.Minutes(5)))  // Order bundling
        .Aggregate(orders => orders.Combine())
        .ToKafka("unified-orders"))
    .OrchestrateLongRunning(async (order) => {
        await inventoryCheckAsync(order);
        await paymentProcessingAsync(order);
        await fulfillmentCoordinationAsync(order);
        await customerNotificationAsync(order);
    });

// Reusable pattern: IoT data processing with ML integration
var manufacturingWorkflow = Temporal.WorkflowBuilder
    .OnKafkaEvent("sensor-data")
    .FlinkProcess(env => env
        .FromKafka("iot-sensors")
        .KeyBy(reading => reading.MachineId)
        .Window(SlidingTimeWindow.Of(Time.Minutes(10), Time.Minutes(1)))
        .Aggregate(readings => readings.CalculateMetrics())
        .Filter(metrics => metrics.AnomalyScore > 0.8)
        .ToKafka("anomaly-alerts"))
    .OrchestrateLongRunning(async (anomaly) => {
        var prediction = await callMLModelAsync(anomaly);
        await scheduleMaintenanceAsync(prediction);
        await notifyTechniciansAsync(anomaly);
        await adjustProductionParametersAsync(prediction);
    });

// Reusable pattern: Critical event processing with care coordination  
var healthcareWorkflow = Temporal.WorkflowBuilder
    .OnKafkaEvent("patient-vitals")
    .FlinkProcess(env => env
        .FromKafka("vital-signs")
        .KeyBy(vital => vital.PatientId)
        .Map(vital => vital.AnalyzeWithAI())  // Real-time AI analysis
        .Filter(analysis => analysis.RequiresIntervention)
        .Rebalance()  // Load balancing for critical alerts
        .ToKafka("critical-alerts"))
    .OrchestrateLongRunning(async (alert) => {
        await notifyNursingStationAsync(alert);
        await escalateToPhysicianAsync(alert);
        await prepareEmergencyProtocolsAsync(alert);
        await updatePatientRecordAsync(alert);
    });

// Reusable pattern: Media processing with real-time engagement
var mediaWorkflow = Temporal.WorkflowBuilder
    .OnKafkaEvent("content-streams")
    .FlinkProcess(env => env
        .FromKafka("live-content")
        .Map(content => content.ExtractMetadata())
        .Filter(content => content.PassesModerationAsync())  // AI content moderation
        .PartitionCustom((content, partitions) => 
            content.ContentType.GetHashCode() % partitions)
        .ToKafka("moderated-content"))
    .OrchestrateLongRunning(async (content) => {
        await generateThumbnailsAsync(content);
        await createSubtitlesAsync(content);  // AI-powered
        await distributeToChannelsAsync(content);
        await trackEngagementMetricsAsync(content);
    });

// Pattern: Real-time AI inference with fallback strategies
var aiWorkflow = Temporal.WorkflowBuilder
    .OnKafkaEvent("inference-requests")
    .FlinkProcess(env => env
        .FromKafka("ai-requests")
        .Map(request => request.PreprocessForModel())
        .KeyBy(request => request.ModelType)  // Route by AI model
        .Rebalance()  // Distribute load across AI workers
        .ToKafka("preprocessed-requests"))
    .OrchestrateLongRunning(async (request) => {
        try {
            var result = await callPrimaryAIModelAsync(request);
            await cacheResultAsync(result);
            return result;
        } catch (ModelUnavailableException) {
            return await callFallbackModelAsync(request);
        }
    });

// Same patterns for CI/CD as business workflows
var buildWorkflow = Temporal.WorkflowBuilder
    .OnKafkaEvent("git-commits")
    .FlinkProcess(env => env
        .FromKafka("code-changes")
        .Filter(change => change.AffectsProduction())
        .Map(change => change.DetermineTestStrategy())
        .ToKafka("build-requests"))
    .OrchestrateLongRunning(async (buildRequest) => {
        await runTestSuiteAsync(buildRequest);
        await buildArtifactsAsync(buildRequest);
        await deployToStagingAsync(buildRequest);
        await runIntegrationTestsAsync(buildRequest);
        await deployToProductionAsync(buildRequest);
    });

var env = Flink.GetExecutionEnvironment();

var dataStream = env.FromCollection(new[] { 1, 2, 3, 4, 5 });

// Rebalance: Uniformly distribute data across all parallel operators
var rebalanced = dataStream
    .Map(x => x * 2)
    .Rebalance()  // Apache Flink 2.0 rebalance operation
    .Filter(x => x > 5);

// Rescale: Distribute to subset of operators (more efficient for different parallelisms)
var rescaled = dataStream
    .Map(x => x * 3)
    .Rescale()   // Apache Flink 2.0 rescale operation
    .Filter(x => x > 10);

// Forward: Direct forwarding (same parallelism required)
var forwarded = dataStream
    .Forward()   // Apache Flink 2.0 forward partitioning
    .Map(x => x + 1);

// Shuffle: Random distribution
var shuffled = dataStream
    .Shuffle()   // Apache Flink 2.0 shuffle partitioning
    .Map(x => x * 2);

// Broadcast: Send to all operators
var broadcasted = dataStream
    .Broadcast() // Apache Flink 2.0 broadcast partitioning
    .Map(x => x + 10);

// Custom partitioning
var customPartitioned = dataStream
    .PartitionCustom(
        (key, numPartitions) => key % numPartitions,  // Custom partitioner
        x => x.GetHashCode()                          // Key selector
    );

await env.ExecuteAsync("Dynamic Partitioning Example");

var env = Flink.GetExecutionEnvironment();

// Configure parallelism and scaling parameters
env.SetParallelism(8)                    // Default parallelism
   .SetMaxParallelism(128)               // Maximum parallelism for scaling
   .EnableAdaptiveScheduler()            // Apache Flink 2.0 adaptive scheduler
   .EnableReactiveMode();                // Apache Flink 2.0 reactive mode

var dataStream = env.FromCollection(data)
    .SetParallelism(4)                   // Operator-specific parallelism
    .SetMaxParallelism(64)              // Operator-specific max parallelism
    .SlotSharingGroup("data-processing") // Fine-grained resource management
    .Map(x => processData(x))
    .Rebalance()                        // Dynamic rebalancing
    .SetParallelism(8);                 // Scale specific operation

await env.ExecuteAsync("Scalable Processing Job");

// Start job from savepoint for scaling
var env = Flink.GetExecutionEnvironment()
    .FromSavepoint("/path/to/savepoint")  // Restore from savepoint
    .SetParallelism(16);                  // New parallelism

// Execute job asynchronously to get JobClient
var jobClient = await env.ExecuteAsyncJob("Scaled Job");

// Trigger savepoint for scaling
var savepointResult = await jobClient.TriggerSavepointAsync("/path/to/new/savepoint");

// Stop job with savepoint for clean scaling
var stopResult = await jobClient.StopWithSavepointAsync("/path/to/scaling/savepoint", drain: true);

// Cancel with savepoint (alternative approach)
var cancelResult = await jobClient.CancelWithSavepointAsync();

// Monitor job status during scaling
var status = await jobClient.GetJobStatusAsync();
Console.WriteLine($"Job {status.JobName}: {status.State}, Parallelism: {status.Parallelism}/{status.MaxParallelism}");

// Modern DataStream API (Apache Flink 2.0 compatible)
var env = Flink.GetExecutionEnvironment();
env.SetParallelism(4)
   .EnableAdaptiveScheduler()
   .EnableReactiveMode();

var dataStream = env.FromCollection(new[] { 1, 2, 3, 4, 5 });
dataStream
    .Map(x => x * 2)
    .Rebalance()           // Rebalance across all operators
    .Filter(x => x > 5)
    .Rescale()             // Rescale to subset
    .Print();

await env.ExecuteAsync("My Job");

// JobBuilder API (Alternative fluent approach)
var job = Flink.JobBuilder
    .FromKafka("orders")
    .Where("Amount > 100")
    .GroupBy("Region")
    .Aggregate("SUM", "Amount")
    .ToKafka("high-value-orders");

await job.Submit("Processing Job");

// Enterprise-scale FlinkDotNet.Orchestration for thousands of clusters
var orchestra = new FlinkOrchestra(logger);

// Provision clusters with auto-scaling
await orchestra.ProvisionClusterAsync(new ClusterConfiguration
{
    Name = "production-cluster",
    TaskSlots = 8,
    TaskManagers = 4
});

// Submit jobs with intelligent placement
var result = await orchestra.SubmitJobAsync(jobDefinition, SubmissionStrategy.BestFit);

// Start Temporal FlinkDotNet.Orchestration workflows
await orchestra.StartOrchestrationWorkflowAsync(new OrchestrationRequest
{
    TargetClusters = 1000,
    MinClusters = 10,
    MaxClusters = 5000
});

var config = new ExecutionConfig()
    .SetParallelism(8)
    .SetMaxParallelism(128)
    .EnableAdaptiveScheduler()           // Apache Flink 2.0 intelligent scheduling
    .EnableReactiveMode()                // Apache Flink 2.0 elastic scaling
    .SetRestartStrategy("exponential-delay")  // Advanced fault tolerance
    .EnableSlotSharing()                 // Resource optimization
    .EnableObjectReuse()                 // Performance optimization
    .SetAutoWatermarkInterval(200);      // Event time processing

var env = Flink.GetExecutionEnvironment(config);

var env = Flink.GetExecutionEnvironment();

// Configure Apache Flink 2.0 features
env.SetParallelism(4)
   .SetMaxParallelism(128)              // Enable dynamic scaling
   .EnableAdaptiveScheduler()           // Automatic parallelism adjustment
   .EnableReactiveMode()                // Adapt to cluster resources
   .EnableCheckpointing(5000);          // Checkpointing for fault tolerance

var numbers = env.FromCollection(Enumerable.Range(1, 1000));

var result = numbers
    .Filter(x => x % 2 == 0)      // Even numbers only
    .Map(x => x * x)              // Square them
    .Rebalance()                  // Apache Flink 2.0 rebalancing
    .SetParallelism(8)            // Scale this operation
    .Sum();                       // Sum the results

await env.ExecuteAsync("Even Squares with Dynamic Scaling");

var env = Flink.GetExecutionEnvironment();
env.EnableAdaptiveScheduler()
   .EnableReactiveMode();

var dataStream = env.FromCollection(generateData());

// Demonstrate all Apache Flink 2.0 partitioning strategies
var processed = dataStream
    .Map(x => processData(x))
    .SetParallelism(4)
    .SlotSharingGroup("data-processing")    // Fine-grained resource management
    
    // Rebalance: Uniform distribution across all operators
    .Rebalance()
    .Map(x => enrichData(x))
    .SetParallelism(8)
    
    // Rescale: Efficient distribution for different parallelisms  
    .Rescale()
    .Filter(x => x.IsValid)
    .SetParallelism(4)
    
    // Forward: Direct forwarding (same parallelism)
    .Forward()
    .Map(x => finalProcessing(x))
    .SetParallelism(4)
    
    // Custom partitioning based on business logic
    .PartitionCustom(
        (key, numPartitions) => key.GetHashCode() % numPartitions,
        x => x.CustomerId
    )
    .SlotSharingGroup("customer-processing");

await env.ExecuteAsync("Advanced Partitioning Example");

var job = Flink.JobBuilder
    .FromKafka("input-topic", config => {
        config.BootstrapServers = "localhost:9092";
        config.GroupId = "processing-group";
    })
    .Map("processed = transform(data)")
    .Where("processed.isValid")
    .ToKafka("output-topic");

// Configure Apache Flink 2.0 features for the job
await job.Configure(config => {
    config.EnableAdaptiveScheduler()
          .EnableReactiveMode()
          .SetParallelism(8)
          .SetMaxParallelism(128);
}).Submit("Kafka Processing with Auto-Scaling");

var job = Flink.JobBuilder
    .FromKafka("events")
    .GroupBy("userId")
    .Window("TUMBLING", 5, "MINUTES")
    .Aggregate("COUNT", "*")
    .ToKafka("user-activity");

await job.Configure(config => {
    config.EnableReactiveMode()          // Adapt to cluster resources
          .SetRestartStrategy("exponential-delay")  // Advanced fault tolerance
          .EnableSlotSharing();          // Resource optimization
}).Submit("User Activity with Reactive Scaling");

var orchestra = new FlinkOrchestra(logger);

// Provision a new cluster with Apache Flink 2.0 features
var cluster = await orchestra.ProvisionClusterAsync(new ClusterConfiguration
{
    Name = "production-west",
    TaskSlots = 16,
    TaskManagers = 8,
    Region = "us-west-2",
    HighAvailability = true,
    AdaptiveSchedulerEnabled = true,    // Enable Apache Flink 2.0 adaptive scheduler
    ReactiveModeEnabled = true          // Enable reactive mode
});

// Get cluster health across all clusters
var health = await orchestra.GetClusterHealthAsync();
Console.WriteLine($"Overall Health Score: {health.OverallHealthScore:F1}%");
Console.WriteLine($"Total Clusters: {health.TotalClusters}");
Console.WriteLine($"Healthy: {health.HealthyClusters}, Critical: {health.CriticalClusters}");

// Define a job with Apache Flink 2.0 configuration
var jobDefinition = new FlinkJobDefinition
{
    JobId = "analytics-pipeline",
    JobName = "Real-time Analytics",
    JobGraph = "...", // Generated from DataStream/JobBuilder
    Parallelism = 8,
    MaxParallelism = 128,               // Enable dynamic scaling
    AdaptiveSchedulerEnabled = true,    // Intelligent resource management
    ReactiveModeEnabled = true,         // Automatic adaptation
    Priority = JobPriority.High
};

// Submit with intelligent placement
var result = await orchestra.SubmitJobAsync(jobDefinition, SubmissionStrategy.BestFit);

if (result.Success)
{
    Console.WriteLine($"Job {result.JobId} submitted to cluster {result.ClusterId}");
    Console.WriteLine($"Flink Job ID: {result.FlinkJobId}");
    
    // Monitor scaling behavior
    var jobClient = result.JobClient;
    var status = await jobClient.GetJobStatusAsync();
    Console.WriteLine($"Current Parallelism: {status.Parallelism}/{status.MaxParallelism}");
}

// Execute job with scaling capabilities
var jobClient = await env.ExecuteAsyncJob("Scalable Analytics Job");

// Monitor and scale using savepoints
var status = await jobClient.GetJobStatusAsync();
Console.WriteLine($"Initial Parallelism: {status.Parallelism}");

// Create savepoint for scaling
var savepointResult = await jobClient.TriggerSavepointAsync("/savepoints/scaling-point");
if (savepointResult.Success)
{
    Console.WriteLine($"Savepoint created at: {savepointResult.SavepointPath}");
    
    // Stop job gracefully for scaling
    var stopResult = await jobClient.StopWithSavepointAsync(savepointPath: savepointResult.SavepointPath, drain: true);
    
    if (stopResult.Success)
    {
        // Restart with new parallelism
        var scaledEnv = Flink.GetExecutionEnvironment()
            .FromSavepoint(stopResult.SavepointPath)    // Restore from savepoint
            .SetParallelism(16)                         // New parallelism
            .SetMaxParallelism(256)                     // New max parallelism
            .EnableAdaptiveScheduler()
            .EnableReactiveMode();
        
        // Re-execute with scaled configuration
        var scaledJobClient = await scaledEnv.ExecuteAsyncJob("Scaled Analytics Job");
        var scaledStatus = await scaledJobClient.GetJobStatusAsync();
        Console.WriteLine($"Scaled Parallelism: {scaledStatus.Parallelism}");
    }
}

// Start long-running orchestration workflow with Apache Flink 2.0 features
var workflowId = await orchestra.StartOrchestrationWorkflowAsync(new OrchestrationRequest
{
    RequestId = "scaling-request-1",
    TargetClusters = 500,
    MinClusters = 50,
    MaxClusters = 2000,
    ScalingPolicy = "demand-based",
    AdaptiveSchedulerEnabled = true,    // Enable intelligent scheduling across clusters
    ReactiveModeEnabled = true          // Enable reactive scaling
});

Console.WriteLine($"Started orchestration workflow: {workflowId}");

// Monitor and scale dynamically
var scalingResult = await orchestra.ScaleOrchestraAsync(targetCapacity: 750);
Console.WriteLine($"Scaled from {scalingResult.PreviousCapacity} to {scalingResult.NewCapacity} clusters");

using Flink.JobBuilder.Backpressure;

// Configure rate limiter with adaptive behavior
var rateLimiter = new TokenBucketRateLimiter(
    rateLimit: 1000.0,      // 1000 operations per second
    burstCapacity: 2000.0   // Handle bursts up to 2000
);

// Use in your application with automatic backpressure handling
if (rateLimiter.TryAcquire())
{
    await ProcessMessage(message);
}
else
{
    // Apache Flink 2.0 handles backpressure automatically
    // This provides additional application-level control
    await Task.Delay(100); // Wait and retry
}

// Configure backpressure in execution environment
var env = Flink.GetExecutionEnvironment();
env.GetConfig()
   .SetProperty("taskmanager.network.memory.max-buffers-per-channel", "10")
   .SetProperty("taskmanager.network.memory.buffers-per-channel", "2")
   .EnableObjectReuse();  // Reduce GC pressure

[Fact]
public async Task TestStreamProcessingWithScaling()
{
    var env = Flink.GetExecutionEnvironment();
    env.EnableAdaptiveScheduler()
       .EnableReactiveMode()
       .SetMaxParallelism(128);
    
    var testData = new[] { 1, 2, 3, 4, 5 };
    var result = env.FromCollection(testData)
        .Map(x => x * 2)
        .Rebalance()                    // Test Apache Flink 2.0 rebalancing
        .SetParallelism(4)              // Test dynamic parallelism
        .CollectAsync();
        
    var expected = new[] { 2, 4, 6, 8, 10 };
    Assert.Equal(expected, await result);
}

[Fact]  
public async Task TestSavepointBasedScaling()
{
    var jobClient = await env.ExecuteAsyncJob("Test Scaling Job");
    
    // Test savepoint creation
    var savepointResult = await jobClient.TriggerSavepointAsync();
    Assert.True(savepointResult.Success);
    
    // Test graceful stopping with savepoint
    var stopResult = await jobClient.StopWithSavepointAsync(drain: true);
    Assert.True(stopResult.Success);
    Assert.True(stopResult.Drained);
}

// LocalTesting/Program.cs
var builder = DistributedApplication.CreateBuilder(args);

var kafka = builder.AddKafka("kafka");

// Apache Flink 2.0 cluster with advanced features
var flink = builder.AddContainer("flink", "flink:2.0-latest")
    .WithEnvironment("FLINK_PROPERTIES", 
        "scheduler-mode: adaptive\n" +
        "scheduler.adaptive.scaling-enabled: true\n" +
        "scheduler.adaptive.resource.wait-timeout: 60s\n" +
        "execution.checkpointing.interval: 5s\n" +
        "parallelism.default: 4\n" +
        "parallelism.default.sink: 8\n" +
        "taskmanager.numberOfTaskSlots: 8");

var gateway = builder.AddProject<Projects.FlinkDotNet_Gateway>("gateway")
    .WithReference(flink);

var testApp = builder.AddProject<Projects.TestApp>("testapp")
    .WithReference(gateway)
    .WithReference(kafka);

builder.Build().Run();

// Enable all Apache Flink 2.0 features
var env = Flink.GetExecutionEnvironment()
    .EnableAdaptiveScheduler()     // Intelligent resource management
    .EnableReactiveMode()          // Elastic scaling
    .SetMaxParallelism(256)        // Dynamic scaling support
    .EnableCheckpointing(5000);    // Enhanced fault tolerance

var scalableStream = env.FromCollection(data)
    .Rebalance()                   // Apache Flink 2.0 rebalancing
    .SetParallelism(8)             // Dynamic parallelism
    .SlotSharingGroup("processing"); // Fine-grained resources

// Complex processing with scaling requirements
var env = Flink.GetExecutionEnvironment()
    .EnableAdaptiveScheduler()
    .EnableReactiveMode();
var stream = env.FromCollection(data)
    .Rebalance()
    .SetParallelism(8);

// Simple pipeline with fluent syntax
var job = Flink.JobBuilder
    .FromKafka("input")
    .Map("process(data)")
    .ToKafka("output");

// Enterprise multi-cluster orchestration
var orchestra = new FlinkOrchestra(logger);
await orchestra.SubmitJobAsync(jobDef, SubmissionStrategy.BestFit);

// Existing code (still works)
var env = Flink.GetExecutionEnvironment();
var stream = env.FromCollection(data).Map(x => x * 2);

// Enhanced with Apache Flink 2.0 features
var enhancedEnv = Flink.GetExecutionEnvironment()
    .EnableAdaptiveScheduler()     // Add intelligent scheduling
    .EnableReactiveMode()          // Add elastic scaling
    .SetMaxParallelism(128);       // Enable dynamic scaling

var enhancedStream = enhancedEnv.FromCollection(data)
    .Map(x => x * 2)
    .Rebalance()                   // Add efficient rebalancing
    .SetParallelism(8);            // Set optimal parallelism