# 🛠️ Resilience Features for KsqlDB Partition Rebalancing

> **Built-in resilience mechanisms for production stability**

This documentation covers the enhanced resilience features in `@gftdcojp/ksqldb-orm` that provide automatic handling of KsqlDB partition rebalancing and other production issues.

## 🎯 Overview

The resilience features address the common production issue: `Cannot determine which host contains the required partitions to serve the pull query` during KsqlDB partition rebalancing.

### Key Features

- ✅ **Built-in retry mechanism** with exponential backoff
- ✅ **Partition-aware error detection**
- ✅ **Circuit breaker pattern** for cascading failure prevention
- ✅ **HTTP/WebSocket fallback** strategies
- ✅ **Backward compatible** API
- ✅ **Metrics collection** for observability
- ✅ **Zero configuration** resilience by default

## 🚀 Quick Start

### Basic Usage (Backward Compatible)

```typescript
import { initializeResilientKsqlDbClient, executePullQueryResilient } from '@gftdcojp/ksqldb-orm';

// Initialize with resilience features (backward compatible)
await initializeResilientKsqlDbClient({
  url: 'https://your-ksqldb-cluster.amazonaws.com:8088',
  apiKey: 'your-api-key',
  apiSecret: 'your-api-secret'
});

// Execute queries - resilience is automatic
const data = await executePullQueryResilient('SELECT * FROM USERS_TABLE;');
```

### Enhanced Configuration

```typescript
await initializeResilientKsqlDbClient({
  url: 'https://your-ksqldb-cluster.amazonaws.com:8088',
  apiKey: 'your-api-key',
  apiSecret: 'your-api-secret',
  resilience: {
    retries: {
      maxRetries: 5,
      baseDelay: 1500,
      backoffStrategy: 'exponential',
      jitter: 0.1
    },
    circuitBreaker: {
      enabled: true,
      failureThreshold: 5,
      openTimeout: 30000
    },
    partitionAwareness: {
      enabled: true,
      rebalanceTimeout: 10000
    },
    fallback: {
      fallbackToHttp: true,
      fallbackTimeout: 5000
    },
    metrics: {
      enabled: true,
      interval: 60000,
      collector: (metrics) => {
        console.log('Resilience metrics:', metrics);
      }
    }
  }
});
```

## 📊 Enhanced API Usage

### Per-Query Configuration

```typescript
// Query with specific resilience settings
const data = await executePullQueryResilient('SELECT * FROM USERS_TABLE;', {
  retries: 3,
  backoffStrategy: 'linear',
  fallbackToHttp: true,
  timeout: 10000
});
```

### Class-Based Client

```typescript
import { ResilientKsqlDbClient } from '@gftdcojp/ksqldb-orm';

const client = new ResilientKsqlDbClient({
  url: 'https://your-ksqldb-cluster.amazonaws.com:8088',
  apiKey: 'your-api-key',
  apiSecret: 'your-api-secret',
  resilience: {
    retries: { maxRetries: 5 },
    circuitBreaker: { enabled: true }
  }
});

await client.initialize();

// Enhanced query execution with detailed result
const result = await client.executePullQuery('SELECT * FROM USERS_TABLE;');

console.log('Data:', result.data);
console.log('Retry count:', result.resilience.retryCount);
console.log('Execution time:', result.resilience.executionTime);
console.log('Fallback used:', result.resilience.fallbackUsed);
console.log('Circuit breaker state:', result.resilience.circuitBreakerState);
```

### Database Client Integration

```typescript
import { createResilientDatabaseClient } from '@gftdcojp/ksqldb-orm';

const db = createResilientDatabaseClient({
  ksql: {
    url: 'https://your-ksqldb-cluster.amazonaws.com:8088',
    apiKey: 'your-api-key',
    apiSecret: 'your-api-secret',
    resilience: {
      retries: { maxRetries: 5 },
      circuitBreaker: { enabled: true }
    }
  },
  schemaRegistry: {
    url: 'https://your-schema-registry.amazonaws.com'
  }
});

// Enhanced query builder
const users = await db.from('users')
  .select('id, name, email')
  .eq('status', 'active')
  .executeEnhanced({
    retries: 3,
    fallbackToHttp: true
  });
```

## ⚙️ Configuration Reference

### Retry Configuration

```typescript
interface RetryConfig {
  /** Maximum number of retries (default: 5) */
  maxRetries?: number;
  /** Base delay between retries in milliseconds (default: 1500) */
  baseDelay?: number;
  /** Maximum delay between retries in milliseconds (default: 30000) */
  maxDelay?: number;
  /** Backoff strategy (default: 'exponential') */
  backoffStrategy?: 'exponential' | 'linear' | 'fixed';
  /** Jitter to add randomness to retry delays (default: 0.1) */
  jitter?: number;
  /** Retry only specific error types */
  retryableErrors?: string[];
}
```

### Circuit Breaker Configuration

```typescript
interface CircuitBreakerConfig {
  /** Whether circuit breaker is enabled (default: true) */
  enabled?: boolean;
  /** Failure threshold to open circuit (default: 5) */
  failureThreshold?: number;
  /** Time window for failure tracking in milliseconds (default: 60000) */
  timeWindow?: number;
  /** Time to wait before attempting to close circuit in milliseconds (default: 30000) */
  openTimeout?: number;
  /** Number of successful calls needed to close circuit (default: 2) */
  successThreshold?: number;
}
```

### Fallback Configuration

```typescript
interface FallbackConfig {
  /** Whether to fallback to HTTP when WebSocket fails (default: true) */
  fallbackToHttp?: boolean;
  /** Alternative endpoints to try */
  alternativeEndpoints?: string[];
  /** Timeout for fallback attempts in milliseconds (default: 5000) */
  fallbackTimeout?: number;
}
```

## 📈 Monitoring & Observability

### Health Checks

```typescript
import { getKsqlDbHealth } from '@gftdcojp/ksqldb-orm';

const health = await getKsqlDbHealth();
console.log('Status:', health.status); // 'healthy' | 'degraded' | 'unhealthy'
console.log('Circuit breaker state:', health.details.circuitBreakerState);
console.log('Metrics:', health.details.metrics);
```

### Metrics Collection

```typescript
import { getKsqlDbMetrics } from '@gftdcojp/ksqldb-orm';

const metrics = getKsqlDbMetrics();
console.log('Total requests:', metrics.totalRequests);
console.log('Success rate:', metrics.successfulRequests / metrics.totalRequests);
console.log('Partition rebalance events:', metrics.partitionRebalanceEvents);
console.log('Average response time:', metrics.averageResponseTime);
```

### Custom Metrics Collector

```typescript
await initializeResilientKsqlDbClient({
  url: '...',
  resilience: {
    metrics: {
      enabled: true,
      interval: 30000, // Collect every 30 seconds
      collector: (metrics) => {
        // Send to your monitoring system
        sendToDatadog(metrics);
        sendToPrometheus(metrics);
        
        // Alerting
        if (metrics.partitionRebalanceEvents > 10) {
          sendAlert('High partition rebalance activity detected');
        }
        
        if (metrics.circuitBreakerTrips > 0) {
          sendAlert('Circuit breaker has been triggered');
        }
      }
    }
  }
});
```

## 🔧 Advanced Usage Patterns

### Environment-Specific Configuration

```typescript
// Production configuration
const productionConfig = {
  url: process.env.KSQLDB_URL,
  apiKey: process.env.KSQLDB_API_KEY,
  apiSecret: process.env.KSQLDB_API_SECRET,
  resilience: {
    retries: {
      maxRetries: 5,
      baseDelay: 2000,
      backoffStrategy: 'exponential' as const
    },
    circuitBreaker: {
      enabled: true,
      failureThreshold: 10,
      openTimeout: 60000
    },
    fallback: {
      fallbackToHttp: true,
      alternativeEndpoints: [
        process.env.KSQLDB_FALLBACK_URL
      ]
    }
  }
};

// Development configuration (more lenient)
const developmentConfig = {
  url: 'http://localhost:8088',
  resilience: {
    retries: { maxRetries: 2 },
    circuitBreaker: { enabled: false },
    metrics: { enabled: false }
  }
};

const config = process.env.NODE_ENV === 'production' 
  ? productionConfig 
  : developmentConfig;

await initializeResilientKsqlDbClient(config);
```

### Migration from Existing Code

```typescript
// Before (existing code)
import { initializeKsqlDbClient, executePullQuery } from '@gftdcojp/ksqldb-orm';

await initializeKsqlDbClient({
  url: 'https://your-cluster.amazonaws.com:8088',
  apiKey: 'your-api-key',
  apiSecret: 'your-api-secret'
});

const data = await executePullQuery('SELECT * FROM USERS_TABLE;');

// After (with resilience - drop-in replacement)
import { initializeResilientKsqlDbClient, executePullQueryResilient } from '@gftdcojp/ksqldb-orm';

await initializeResilientKsqlDbClient({
  url: 'https://your-cluster.amazonaws.com:8088',
  apiKey: 'your-api-key',
  apiSecret: 'your-api-secret'
  // Resilience is enabled by default with sensible defaults
});

const data = await executePullQueryResilient('SELECT * FROM USERS_TABLE;');
```

### Error Handling Patterns

```typescript
import { ResilientKsqlDbClient, ResilienceErrorType } from '@gftdcojp/ksqldb-orm';

const client = new ResilientKsqlDbClient(config);

try {
  const result = await client.executePullQuery('SELECT * FROM USERS_TABLE;');
  
  if (result.error) {
    // Handle partial success scenarios
    console.warn('Query succeeded with resilience features', {
      retryCount: result.resilience.retryCount,
      fallbackUsed: result.resilience.fallbackUsed
    });
  }
  
  return result.data;
} catch (error) {
  // Handle complete failures
  console.error('Query failed even with resilience', error);
  
  // Check circuit breaker state
  if (client.getCircuitBreakerState() === 'OPEN') {
    console.log('Circuit breaker is open - service may be down');
    // Implement fallback logic or user notification
  }
  
  throw error;
}
```

### Push Query Resilience

```typescript
import { ResilientKsqlDbClient } from '@gftdcojp/ksqldb-orm';

const client = new ResilientKsqlDbClient(config);

const { terminate } = await client.executePushQuery(
  'SELECT * FROM USERS_STREAM EMIT CHANGES;',
  (data) => {
    console.log('Received data:', data);
  },
  (error, resilience) => {
    console.error('Push query error:', error.message);
    console.log('Error type:', resilience?.errorType);
    console.log('Can retry:', resilience?.canRetry);
    
    if (resilience?.canRetry) {
      // Implement reconnection logic
      setTimeout(() => {
        console.log('Attempting to reconnect...');
        // Restart push query
      }, 5000);
    }
  }
);
```

## 🧪 Testing with Resilience Features

### Unit Testing

```typescript
import { describe, it, expect, vi } from 'vitest';
import { ResilientKsqlDbClient } from '@gftdcojp/ksqldb-orm';

describe('Resilient KsqlDB Client', () => {
  it('should handle partition rebalancing errors', async () => {
    const client = new ResilientKsqlDbClient({
      url: 'http://localhost:8088',
      resilience: {
        retries: { maxRetries: 2, baseDelay: 100 }
      }
    });

    // Mock the underlying client to simulate partition errors
    vi.spyOn(client as any, 'executePullQuery')
      .mockRejectedValueOnce(new Error('Cannot determine which host contains the required partitions'))
      .mockResolvedValue({ data: [{ id: 1, name: 'test' }] });

    const result = await client.executePullQuery('SELECT * FROM test;');
    
    expect(result.data).toBeDefined();
    expect(result.resilience.retryCount).toBe(1);
    expect(result.error).toBeUndefined();
  });
});
```

### Integration Testing

```typescript
// Test circuit breaker behavior
describe('Circuit Breaker Integration', () => {
  it('should open circuit after repeated failures', async () => {
    const client = new ResilientKsqlDbClient({
      url: 'http://localhost:8088',
      resilience: {
        circuitBreaker: { failureThreshold: 2 }
      }
    });

    // Simulate service down
    vi.spyOn(client as any, 'executePullQuery')
      .mockRejectedValue(new Error('Service unavailable'));

    // First two failures should trip circuit
    await client.executePullQuery('SELECT 1;');
    await client.executePullQuery('SELECT 1;');
    
    expect(client.getCircuitBreakerState()).toBe('OPEN');
    
    // Next query should be rejected immediately
    const result = await client.executePullQuery('SELECT 1;');
    expect(result.error.message).toContain('Circuit breaker is OPEN');
  });
});
```

## 🚨 Production Recommendations

### Monitoring Setup

1. **Set up alerting** for high partition rebalance events:
   ```typescript
   collector: (metrics) => {
     if (metrics.partitionRebalanceEvents > 50) {
       sendAlert('Excessive partition rebalancing detected');
     }
   }
   ```

2. **Monitor circuit breaker state**:
   ```typescript
   const health = await getKsqlDbHealth();
   if (health.details.circuitBreakerState === 'OPEN') {
     sendAlert('KsqlDB circuit breaker is OPEN');
   }
   ```

3. **Track success rates**:
   ```typescript
   const successRate = metrics.successfulRequests / metrics.totalRequests;
   if (successRate < 0.95) {
     sendAlert(`Low success rate: ${successRate * 100}%`);
   }
   ```

### Performance Tuning

```typescript
// High-throughput configuration
const highThroughputConfig = {
  resilience: {
    retries: {
      maxRetries: 3, // Lower for faster failure
      baseDelay: 500, // Shorter delays
      backoffStrategy: 'linear'
    },
    circuitBreaker: {
      failureThreshold: 20, // Higher threshold
      openTimeout: 15000 // Faster recovery
    },
    metrics: {
      interval: 30000 // More frequent collection
    }
  }
};

// High-reliability configuration
const highReliabilityConfig = {
  resilience: {
    retries: {
      maxRetries: 10, // More retries
      baseDelay: 2000, // Longer delays
      backoffStrategy: 'exponential'
    },
    circuitBreaker: {
      failureThreshold: 5, // Lower threshold
      openTimeout: 60000 // Longer recovery time
    },
    fallback: {
      fallbackToHttp: true,
      alternativeEndpoints: ['backup-cluster-url']
    }
  }
};
```

### Deployment Strategies

1. **Blue-Green Deployment** with resilience:
   ```typescript
   // Gradually switch traffic with fallback
   const config = {
     resilience: {
       fallback: {
         alternativeEndpoints: [
           process.env.BLUE_CLUSTER_URL,
           process.env.GREEN_CLUSTER_URL
         ]
       }
     }
   };
   ```

2. **Canary Deployment** monitoring:
   ```typescript
   // Monitor metrics during canary deployment
   const metrics = getKsqlDbMetrics();
   if (metrics.errorRate > 0.05) {
     // Rollback canary
     rollbackDeployment();
   }
   ```

## 🆚 Comparison with Manual Solutions

| Feature | Manual Implementation | Built-in Resilience |
|---------|----------------------|---------------------|
| **Retry Logic** | 50+ lines of code | Zero configuration |
| **Error Classification** | Complex pattern matching | Automatic detection |
| **Circuit Breaker** | Custom state management | Built-in with metrics |
| **Monitoring** | Manual instrumentation | Automatic collection |
| **Maintenance** | Ongoing updates needed | Maintained by library |
| **Testing** | Mock complex scenarios | Testable components |

## 🔗 Related Documentation

- [KsqlDB Client Documentation](./ksqldb-client.md)
- [Query Builder Guide](./query-builder.md)
- [Security Features](../src/security.ts)
- [Testing Guide](../src/__tests__/)

## 📞 Support

If you encounter issues with the resilience features:

1. Check the [health endpoint](#health-checks)
2. Review [metrics](#metrics-collection) for patterns
3. Enable debug logging
4. Refer to the [test cases](../src/__tests__/resilience.test.ts) for examples

For production issues, ensure you have metrics collection enabled and provide the collected data when reporting issues.