69 Abschließende Empfehlungen: Build small, react early, stay decoupled

// Phase 1: Einfache Event-Notification
@Service
public class OrderServiceV1 {
    
    @Autowired
    private OrderRepository orderRepository;
    
    @Autowired
    private ApplicationEventPublisher eventPublisher;
    
    public Order createOrder(CreateOrderRequest request) {
        // Traditionelle Geschäftslogik
        Order order = new Order(request);
        order = orderRepository.save(order);
        
        // Einfache Event-Benachrichtigung hinzufügen
        eventPublisher.publishEvent(new OrderCreatedEvent(order.getId()));
        
        return order;
    }
}

// Phase 2: Externe Event-Publikation
@Service  
public class OrderServiceV2 {
    
    @Autowired
    private KafkaTemplate<String, Object> kafkaTemplate;
    
    public Order createOrder(CreateOrderRequest request) {
        Order order = new Order(request);
        order = orderRepository.save(order);
        
        // Zu externem Event-System erweitert
        OrderPlacedEvent event = OrderPlacedEvent.builder()
                .orderId(order.getId())
                .customerId(order.getCustomerId())
                .totalAmount(order.getTotalAmount())
                .build();
                
        kafkaTemplate.send("order.placed.v1", event);
        
        return order;
    }
}

// Phase 3: Vollständige Event-Driven Implementierung
@Service
public class OrderServiceV3 {
    
    public CompletableFuture<OrderResult> createOrder(CreateOrderRequest request) {
        // Command-Handler Approach
        CreateOrderCommand command = new CreateOrderCommand(request);
        return commandBus.send(command);
    }
    
    @EventHandler
    public void on(OrderCreatedEvent event) {
        // Event-basierte Nachverarbeitung
        publishOrderPlacedEvent(event);
        updateInventoryProjection(event);
        triggerPaymentProcess(event);
    }
}

# Schlecht: Technisch orientierte Events
class DatabaseUpdateEvent:
    table: str
    operation: str  # INSERT, UPDATE, DELETE
    record_id: str

class ServiceCallEvent:
    service_name: str
    method_name: str
    parameters: dict

# Gut: Fachlich orientierte Events  
class OrderPlacedEvent:
    order_id: str
    customer_id: str
    items: List[OrderItem]
    total_amount: Decimal
    placed_at: datetime
    
class PaymentProcessedEvent:
    order_id: str
    payment_id: str
    amount: Decimal
    payment_method: str
    processed_at: datetime

class InventoryReservedEvent:
    order_id: str
    reservations: List[InventoryReservation]
    reserved_at: datetime
    expires_at: datetime

// Order Context - Public Interface Event
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
public class OrderPlacedEvent implements PublicDomainEvent {
    private String orderId;
    private String customerId;
    private BigDecimal totalAmount;
    private String currency;
    private Instant placedAt;
    
    // Keine internen Order-Details wie Berechnungslogik, Validierungsregeln etc.
}

// Payment Context - Internal Event
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
class PaymentAuthorizationRequestedEvent implements InternalDomainEvent {
    private String paymentId;
    private String orderId;
    private PaymentMethod paymentMethod;
    private RiskAssessment riskAssessment;  // Interne Payment-Details
    private AuthorizationStrategy strategy;  // Interne Payment-Logik
}

// Cross-Context Integration Event
@JsonTypeInfo(use = JsonTypeInfo.Id.NAME)
public class PaymentRequiredEvent implements IntegrationEvent {
    private String orderId;
    private BigDecimal amount;
    private String currency;
    private String customerId;
    private Instant requiredBy;
    
    // Nur das Minimum für andere Contexts
}

from enum import Enum
from typing import Optional, Dict, Any
from dataclasses import dataclass

class ErrorRecoveryStrategy(Enum):
    RETRY = "retry"
    DEAD_LETTER = "dead_letter" 
    IGNORE = "ignore"
    COMPENSATE = "compensate"
    ESCALATE = "escalate"

@dataclass
class ErrorHandlingPolicy:
    error_type: str
    strategy: ErrorRecoveryStrategy
    max_retries: int = 3
    retry_delay_seconds: int = 30
    escalation_threshold: int = 10

class EventErrorHandler:
    def __init__(self):
        self.policies = {
            'ValidationError': ErrorHandlingPolicy('ValidationError', ErrorRecoveryStrategy.DEAD_LETTER),
            'NetworkTimeout': ErrorHandlingPolicy('NetworkTimeout', ErrorRecoveryStrategy.RETRY, max_retries=5),
            'DatabaseConstraintViolation': ErrorHandlingPolicy('DatabaseConstraintViolation', ErrorRecoveryStrategy.DEAD_LETTER),
            'ExternalServiceUnavailable': ErrorHandlingPolicy('ExternalServiceUnavailable', ErrorRecoveryStrategy.RETRY, max_retries=10),
            'BusinessRuleViolation': ErrorHandlingPolicy('BusinessRuleViolation', ErrorRecoveryStrategy.COMPENSATE)
        }
    
    async def handle_processing_error(self, event: Dict[str, Any], error: Exception) -> bool:
        """Handle event processing error based on error type"""
        
        error_type = type(error).__name__
        policy = self.policies.get(error_type, 
                                 ErrorHandlingPolicy('Unknown', ErrorRecoveryStrategy.ESCALATE))
        
        if policy.strategy == ErrorRecoveryStrategy.RETRY:
            return await self.retry_processing(event, error, policy)
        elif policy.strategy == ErrorRecoveryStrategy.DEAD_LETTER:
            await self.send_to_dead_letter(event, error)
            return False
        elif policy.strategy == ErrorRecoveryStrategy.COMPENSATE:
            await self.trigger_compensation(event, error)
            return False
        elif policy.strategy == ErrorRecoveryStrategy.ESCALATE:
            await self.escalate_error(event, error)
            return False
        else:
            return False  # IGNORE

@Component
public class EventContractGovernance {
    
    public ContractReviewResult reviewSchemaChange(SchemaChangeRequest request) {
        List<ContractViolation> violations = new ArrayList<>();
        
        // 1. Backward Compatibility Check
        if (!isBackwardCompatible(request.getCurrentSchema(), request.getNewSchema())) {
            violations.add(new ContractViolation(
                ViolationType.BREAKING_CHANGE,
                "Schema change breaks backward compatibility",
                Severity.HIGH
            ));
        }
        
        // 2. Consumer Impact Assessment
        List<String> affectedConsumers = findAffectedConsumers(request.getEventType());
        for (String consumer : affectedConsumers) {
            ConsumerImpact impact = assessConsumerImpact(consumer, request.getNewSchema());
            if (impact.requiresAction()) {
                violations.add(new ContractViolation(
                    ViolationType.CONSUMER_IMPACT,
                    "Consumer " + consumer + " requires changes: " + impact.getDescription(),
                    impact.getSeverity()
                ));
            }
        }
        
        // 3. Governance Rules Check
        violations.addAll(checkGovernanceRules(request));
        
        return ContractReviewResult.builder()
                .approved(violations.stream().noneMatch(v -> v.getSeverity() == Severity.HIGH))
                .violations(violations)
                .requiredApprovals(determineRequiredApprovals(violations))
                .build();
    }
}

class EventStormingSession:
    """Structure event storming sessions for team alignment"""
    
    def __init__(self, domain_area: str, stakeholder_teams: List[str]):
        self.domain_area = domain_area
        self.stakeholder_teams = stakeholder_teams
        self.discovered_events = []
        self.identified_aggregates = []
        self.external_systems = []
    
    def run_session(self) -> EventStormingResult:
        """Run collaborative event discovery session"""
        
        # Phase 1: Event Discovery (Orange Stickies)
        domain_events = self.discover_domain_events()
        
        # Phase 2: Timeline Creation
        event_timeline = self.create_event_timeline(domain_events)
        
        # Phase 3: Command Identification (Blue Stickies)
        commands = self.identify_commands(event_timeline)
        
        # Phase 4: Aggregate Discovery (Yellow Stickies)
        aggregates = self.discover_aggregates(commands, domain_events)
        
        # Phase 5: Bounded Context Identification
        bounded_contexts = self.identify_bounded_contexts(aggregates)
        
        # Phase 6: External System Integration (Pink Stickies)
        external_integrations = self.identify_external_systems(event_timeline)
        
        return EventStormingResult(
            domain_events=domain_events,
            timeline=event_timeline,
            commands=commands,
            aggregates=aggregates,
            bounded_contexts=bounded_contexts,
            external_systems=external_integrations
        )

@Component
public class TechnologyDecisionFramework {
    
    public TechnologyRecommendation recommendEventTechnology(ProjectRequirements requirements) {
        
        // Assess current requirements
        ThroughputLevel throughput = assessThroughput(requirements.getExpectedEventsPerSecond());
        ConsistencyLevel consistency = assessConsistencyNeeds(requirements.getConsistencyRequirements());
        ComplexityTolerance complexity = assessComplexityTolerance(requirements.getTeamExperience());
        
        // Decision matrix
        if (throughput == ThroughputLevel.LOW && complexity == ComplexityTolerance.LOW) {
            return TechnologyRecommendation.builder()
                    .primary("Spring ApplicationEventPublisher")
                    .backup("RabbitMQ")
                    .reasoning("Start simple, upgrade when needed")
                    .migrationPath("ApplicationEventPublisher → RabbitMQ → Kafka")
                    .build();
        }
        
        if (throughput == ThroughputLevel.MEDIUM && consistency == ConsistencyLevel.EVENTUAL) {
            return TechnologyRecommendation.builder()
                    .primary("Apache Kafka")
                    .backup("AWS Kinesis")
                    .reasoning("Good balance of throughput and operational complexity")
                    .considerations("Requires investment in Kafka operations knowledge")
                    .build();
        }
        
        if (throughput == ThroughputLevel.HIGH && requirements.requiresStreamProcessing()) {
            return TechnologyRecommendation.builder()
                    .primary("Apache Kafka + Kafka Streams")
                    .backup("Apache Pulsar + Apache Flink")
                    .reasoning("High-performance streaming with processing capabilities")
                    .considerations("Significant operational overhead, dedicated platform team recommended")
                    .build();
        }
        
        // Default conservative recommendation
        return TechnologyRecommendation.builder()
                .primary("RabbitMQ")
                .reasoning("Proven, well-understood, good operational tooling")
                .migrationPath("Can upgrade to Kafka when throughput requirements increase")
                .build();
    }
}

from abc import ABC, abstractmethod
from typing import Dict, Any, List, Callable

class EventPublisher(ABC):
    """Abstract event publisher to avoid vendor lock-in"""
    
    @abstractmethod
    async def publish(self, topic: str, event: Dict[str, Any]) -> bool:
        pass
    
    @abstractmethod
    async def publish_batch(self, topic: str, events: List[Dict[str, Any]]) -> List[bool]:
        pass

class KafkaEventPublisher(EventPublisher):
    def __init__(self, bootstrap_servers: str):
        self.producer = KafkaProducer(bootstrap_servers=bootstrap_servers)
    
    async def publish(self, topic: str, event: Dict[str, Any]) -> bool:
        try:
            future = self.producer.send(topic, value=event)
            result = future.get(timeout=10)
            return True
        except Exception:
            return False

class AWSEventBridgePublisher(EventPublisher):
    def __init__(self, region: str):
        self.eventbridge = boto3.client('events', region_name=region)
    
    async def publish(self, topic: str, event: Dict[str, Any]) -> bool:
        try:
            response = self.eventbridge.put_events(
                Entries=[{
                    'Source': event.get('source', 'application'),
                    'DetailType': event.get('type', 'Custom Event'),
                    'Detail': json.dumps(event.get('data', {})),
                    'EventBusName': topic
                }]
            )
            return response['FailedEntryCount'] == 0
        except Exception:
            return False

class EventPublisherFactory:
    """Factory to create appropriate publisher based on configuration"""
    
    @staticmethod
    def create_publisher(config: Dict[str, Any]) -> EventPublisher:
        provider = config.get('provider', 'kafka')
        
        if provider == 'kafka':
            return KafkaEventPublisher(config['bootstrap_servers'])
        elif provider == 'aws_eventbridge':
            return AWSEventBridgePublisher(config['region'])
        elif provider == 'azure_eventhub':
            return AzureEventHubPublisher(config['connection_string'])
        else:
            raise ValueError(f"Unsupported event publisher provider: {provider}")

@Service
public class OrderServiceMigration {
    
    @Value("${migration.events.enabled:false}")
    private boolean eventsEnabled;
    
    @Value("${migration.events.percentage:0}")
    private int eventPercentage;
    
    private final Random random = new Random();
    
    public Order createOrder(CreateOrderRequest request) {
        Order order = createOrderCore(request);
        
        // Legacy synchronous processing
        if (!eventsEnabled || random.nextInt(100) >= eventPercentage) {
            processOrderSynchronously(order);
        } else {
            // New event-driven processing
            processOrderAsynchronously(order);
        }
        
        return order;
    }
    
    private void processOrderSynchronously(Order order) {
        // Legacy approach
        paymentService.processPayment(order);
        inventoryService.reserveInventory(order);
        shippingService.createShipment(order);
    }
    
    private void processOrderAsynchronously(Order order) {
        // New event-driven approach
        OrderPlacedEvent event = OrderPlacedEvent.builder()
                .orderId(order.getId())
                .customerId(order.getCustomerId())
                .items(order.getItems())
                .build();
                
        eventPublisher.publish("order.placed.v1", event);
    }
}

from typing import Dict, Any
import asyncio

class FeatureFlaggedEventProcessor:
    """Use feature flags to control EDA evolution"""
    
    def __init__(self, feature_flag_client):
        self.feature_flags = feature_flag_client
    
    async def process_order(self, order_data: Dict[str, Any]):
        """Process order with feature-flag controlled behavior"""
        
        # Check feature flags
        async_processing_enabled = await self.feature_flags.is_enabled(
            'async_order_processing', 
            context={'customer_tier': order_data.get('customerTier')}
        )
        
        event_sourcing_enabled = await self.feature_flags.is_enabled(
            'order_event_sourcing',
            context={'order_value': order_data.get('totalAmount', 0)}
        )
        
        saga_pattern_enabled = await self.feature_flags.is_enabled(
            'order_fulfillment_saga'
        )
        
        # Process based on enabled features
        if event_sourcing_enabled:
            await self.process_with_event_sourcing(order_data)
        elif async_processing_enabled:
            await self.process_with_events(order_data)
        else:
            await self.process_synchronously(order_data)
        
        # Optional saga initiation
        if saga_pattern_enabled and order_data.get('totalAmount', 0) > 1000:
            await self.initiate_saga(order_data)
    
    async def process_with_event_sourcing(self, order_data: Dict[str, Any]):
        """Event sourcing approach"""
        commands = [
            CreateOrderCommand(order_data),
            ValidateOrderCommand(order_data['orderId']),
            ReserveInventoryCommand(order_data['orderId'], order_data['items'])
        ]
        
        for command in commands:
            await self.command_bus.send(command)
    
    async def process_with_events(self, order_data: Dict[str, Any]):
        """Simple event-driven approach"""
        event = OrderPlacedEvent(
            order_id=order_data['orderId'],
            customer_id=order_data['customerId'],
            items=order_data['items']
        )
        
        await self.event_bus.publish('order.placed.v1', event)
    
    async def process_synchronously(self, order_data: Dict[str, Any]):
        """Legacy synchronous approach"""
        await self.payment_service.process_payment(order_data)
        await self.inventory_service.reserve_items(order_data['items'])
        await self.shipping_service.create_shipment(order_data)

@Component
public class EDAEvolutionMetrics {
    
    private final MeterRegistry meterRegistry;
    
    @EventListener
    public void onProcessingCompleted(ProcessingCompletedEvent event) {
        // Track processing patterns
        Timer.Sample sample = Timer.start(meterRegistry);
        sample.stop(Timer.builder("event.processing.duration")
                .tag("processing.type", event.getProcessingType())
                .tag("event.type", event.getEventType())
                .register(meterRegistry));
        
        // Track success rates by processing type
        Counter.builder("event.processing.completed")
                .tag("processing.type", event.getProcessingType())
                .tag("success", String.valueOf(event.isSuccessful()))
                .register(meterRegistry)
                .increment();
    }
    
    @Scheduled(fixedDelay = 300000) // Every 5 minutes
    public void analyzeEvolutionOpportunities() {
        
        // Analyze current performance patterns
        EvolutionAnalysis analysis = performEvolutionAnalysis();
        
        if (analysis.shouldRecommendEventDriven()) {
            log.info("Recommendation: Consider migrating {} to event-driven processing", 
                    analysis.getCandidateServices());
            
            // Create evolution recommendation
            createEvolutionRecommendation(analysis);
        }
    }
    
    private EvolutionAnalysis performEvolutionAnalysis() {
        // Get metrics for last 24 hours
        Duration lookback = Duration.ofDays(1);
        
        // Analyze synchronous processing bottlenecks
        Map<String, Double> syncProcessingLatencies = getSyncProcessingLatencies(lookback);
        Map<String, Double> errorRates = getErrorRates(lookback);
        Map<String, Double> throughputLimits = getThroughputLimits(lookback);
        
        // Find candidates for event-driven migration
        List<String> candidateServices = syncProcessingLatencies.entrySet().stream()
                .filter(entry -> entry.getValue() > 5000) // > 5 seconds
                .filter(entry -> errorRates.get(entry.getKey()) > 0.05) // > 5% error rate
                .map(Map.Entry::getKey)
                .collect(Collectors.toList());
        
        return EvolutionAnalysis.builder()
                .candidateServices(candidateServices)
                .currentBottlenecks(syncProcessingLatencies)
                .errorRates(errorRates)
                .recommendEventDriven(!candidateServices.isEmpty())
                .build();
    }
}