The KafkaConsumerObservationHandler is a Spring Boot component that provides distributed tracing for Kafka message consumption in the Wallet Hub application. It creates detailed observability spans for every Kafka consumer operation, enabling end-to-end trace visibility across event-driven workflows.

• Consumer Lag Tracking: Measures time between message production and consumption
• Processing Metrics: Captures deserialization and processing durations
• Error Diagnostics: Records detailed error context for troubleshooting
• Trace Continuity: Maintains distributed trace context across service boundaries

• Low Overhead: <1ms per message impact
• Feature Flag Control: Can be disabled via tracing.features.kafka
• Efficient Attributes: Uses low-cardinality attributes for optimal performance
• Async Export: Spans exported asynchronously to tracing backends

• Spring Cloud Stream: Native integration with Spring's messaging abstraction
• OpenTelemetry Compliance: Follows semantic conventions for messaging systems
• CloudEvents Support: Works with CloudEvent trace context propagation
• Multi-Backend Export: Supports Tempo (OTLP) and Zipkin backends

# application.yml
tracing:
  features:
    kafka: true  # Enable Kafka consumer tracing
    
management:
  tracing:
    enabled: true
  zipkin:
    tracing:
      endpoint: http://localhost:9411/api/v2/spans

Check application logs for:

Started CONSUMER span for Kafka receive [topic=wallet-events, partition=0, offset=42, group=wallet-service]

Access tracing backends:

• Tempo: http://localhost:3000 (if using Grafana)
• Zipkin: http://localhost:9411

-- In tracing backend, query for:
-- messaging.consumer_lag_ms > 1000
-- messaging.processing_time_ms > 500

-- Find failed consumer spans:
-- status = "error"
-- error.type = "ValidationException"

-- Analyze consumer performance:
-- P99(messaging.processing_time_ms)
-- AVG(messaging.consumer_lag_ms)
-- COUNT(*) by messaging.destination.name

• Verify tracing.features.kafka=true
• Check Spring Cloud Stream configuration
• Verify ObservationRegistry bean creation
• Check tracing backend connectivity

• Ensure producers use CloudEventTracePropagator
• Verify CloudEvent has traceparent extension
• Check producer tracing configuration

• Review messaging.processing_time_ms
• Check consumer business logic
• Monitor system resources
• Consider increasing consumer instances

• Monitor messaging.consumer_lag_ms
• Check producer rate vs consumer rate
• Review partition assignment
• Consider consumer scaling

@Component
public class WalletEventConsumer {
    
    @Bean
    public Consumer<Message<CloudEvent>> consumeWalletEvents(
            CloudEventTracePropagator tracePropagator) {
        
        return message -> {
            CloudEvent event = message.getPayload();
            Span span = tracePropagator.extractTraceContext(event);
            
            try {
                // Process event
                processWalletEvent(event);
            } catch (Exception e) {
                span.error(e);
                throw e;
            } finally {
                span.end();
            }
        };
    }
}

// The handler automatically adds standard attributes
// For custom business attributes, use SpanAttributeBuilder:

@Autowired
private SpanAttributeBuilder spanAttributeBuilder;

// In consumer logic:
spanAttributeBuilder.addWalletOperationAttributes(
    span, walletId, "funds_added", transactionId, amount, currency);

1. Enable Sampling: Set management.tracing.sampling.probability=0.01 (1%)
2. Disable Non-Critical: Turn off tracing for low-value topics
3. Monitor Memory: Watch for span accumulation during backpressure
4. Use Async Processing: Ensure span export doesn't block consumer threads

1. Disable Tracing: Set tracing.features.kafka=false for critical paths
2. Minimal Attributes: Use only essential span attributes
3. Batch Processing: Aggregate spans for batch consumers
4. Local Testing: Profile overhead in test environment

• Full Documentation: Comprehensive module documentation
• Tracing Architecture: Overall tracing infrastructure
• CloudEvents Integration: Trace context propagation details

• Health Check: GET /actuator/health/tracing
• Configuration: GET /actuator/configprops/tracing.features
• Metrics: GET /actuator/metrics/observation

# Example Prometheus alert rules
- alert: HighConsumerLag
  expr: messaging_consumer_lag_ms > 1000
  for: 5m
  
- alert: ConsumerErrorRateHigh
  expr: rate(messaging_errors_total[5m]) > 0.1

• Added CloudEvents trace context support
• Improved consumer lag calculation
• Added feature flag configuration
• Enhanced error handling

• Dynamic sampling based on message rate
• Batch consumer support
• Custom attribute injection API
• Enhanced metrics integration

Next Steps: For detailed implementation, configuration, and troubleshooting, refer to the full documentation.