enhancement: Exception handling improvements

[kitalkuyo-gita]

The current codebase for GEAFlow still contains a significant number of exceptions and errors. Improving these errors will help provide a better user experience when exceptions are thrown.

The following areas can be prioritized for improvement.

• ErrorApiResponse.java: Exception classification logic can be more refined.
• ComponentUncaughtExceptionHandler.java: Exception handling can be more elegant.
• SliceOutputChannelHandler.java: Exception handling can be more specific.

[kitalkuyo-gita]

@yazong Are you willing to solve this problem?

[yazong]

OK

[kitalkuyo-gita]

@yazhong I've added an implementation document, which you might find useful.

Overview

improvements for exception handling across three critical components in the GeaFlow system, focusing on enhanced classification, graceful handling, and specific error contexts.

Target Components

1. ErrorApiResponse.java - API exception classification logic
2. ComponentUncaughtExceptionHandler.java - Thread exception handling
3. SliceOutputChannelHandler.java - Network channel exception handling

---

1. ErrorApiResponse Enhancement

Current Issues

• Limited Exception Classification: Only 7 exception types with basic if-else chain
• No Exception Metadata: Missing context like timestamp, request ID, stack trace
• No Retry Guidance: Users don't know if errors are retryable
• Missing Category Mapping: No grouping for similar error types

Solution: Hierarchical Exception Taxonomy

1.1 Create Exception Category Enum

public enum ExceptionCategory {
    // Client errors (4xx equivalent)
    VALIDATION_ERROR(400, "VALIDATION", true, false),
    AUTHENTICATION_ERROR(401, "AUTH", true, false),
    AUTHORIZATION_ERROR(403, "FORBIDDEN", true, false),
    NOT_FOUND_ERROR(404, "NOT_FOUND", true, false),
    
    // Business logic errors (4xx)
    ILLEGAL_STATE_ERROR(409, "ILLEGAL_STATE", true, false),
    COMPILATION_ERROR(422, "COMPILE", true, true),
    
    // Server errors (5xx equivalent)
    INTERNAL_ERROR(500, "INTERNAL", false, true),
    SERVICE_UNAVAILABLE(503, "UNAVAILABLE", false, true),
    TIMEOUT_ERROR(504, "TIMEOUT", false, true),
    
    // Infrastructure errors
    NETWORK_ERROR(502, "NETWORK", false, true),
    DATABASE_ERROR(500, "DATABASE", false, true),
    RESOURCE_EXHAUSTED(503, "RESOURCE", false, true);

    private final int httpStatus;
    private final String code;
    private final boolean clientError;
    private final boolean retryable;
}

1.2 Enhanced ErrorApiResponse

@Getter
public class ErrorApiResponse<T> extends GeaflowApiResponse<T> {
    private final GeaflowApiRequest<?> request;
    private final String message;
    private final String errorCode;
    private final ExceptionCategory category;
    private final long timestamp;
    private final String requestId;
    private final boolean retryable;
    private final String stackTraceId; // Reference to detailed logs
    
    protected ErrorApiResponse(Throwable error) {
        super(false);
        
        // Enhanced classification
        ExceptionClassifier classifier = new ExceptionClassifier();
        ExceptionMetadata metadata = classifier.classify(error);
        
        this.category = metadata.getCategory();
        this.errorCode = metadata.getErrorCode();
        this.code = GeaflowApiResponseCode.fromCategory(category);
        this.message = extractUserFriendlyMessage(error, metadata);
        this.retryable = metadata.isRetryable();
        this.timestamp = System.currentTimeMillis();
        this.requestId = ContextHolder.get().getRequestId();
        this.stackTraceId = logDetailedError(error, metadata);
        this.request = GeaflowApiRequest.currentRequest();
    }
    
    private String extractUserFriendlyMessage(Throwable error, ExceptionMetadata metadata) {
        if (error instanceof GeaflowCompileException) {
            return ((GeaflowCompileException) error).getDisplayMessage();
        }
        if (error instanceof GeaflowException && metadata.isClientError()) {
            return error.getMessage(); // Safe to expose
        }
        // Mask internal errors
        return String.format("%s: %s", category.name(), 
            "Please contact support with error ID: " + stackTraceId);
    }
}

1.3 Exception Classifier

public class ExceptionClassifier {
    private final Map<Class<? extends Throwable>, ExceptionCategory> mappings;
    
    public ExceptionClassifier() {
        mappings = new HashMap<>();
        // Security exceptions
        mappings.put(GeaflowSecurityException.class, AUTHORIZATION_ERROR);
        
        // Validation exceptions
        mappings.put(GeaflowIllegalException.class, VALIDATION_ERROR);
        mappings.put(IllegalArgumentException.class, VALIDATION_ERROR);
        mappings.put(NullPointerException.class, VALIDATION_ERROR);
        
        // Compilation exceptions
        mappings.put(GeaflowCompileException.class, COMPILATION_ERROR);
        
        // Network/Infrastructure
        mappings.put(IOException.class, NETWORK_ERROR);
        mappings.put(TimeoutException.class, TIMEOUT_ERROR);
        
        // Resource errors
        mappings.put(GeaflowOutOfMemoryException.class, RESOURCE_EXHAUSTED);
        
        // Default
        mappings.put(GeaflowException.class, INTERNAL_ERROR);
    }
    
    public ExceptionMetadata classify(Throwable error) {
        // Check exact match
        ExceptionCategory category = mappings.get(error.getClass());
        
        // Check inheritance chain
        if (category == null) {
            category = findByInheritance(error);
        }
        
        // Analyze root cause
        Throwable rootCause = getRootCause(error);
        ExceptionLevel level = determineLevel(category, rootCause);
        
        return ExceptionMetadata.builder()
            .category(category)
            .errorCode(generateErrorCode(category, error))
            .level(level)
            .retryable(category.isRetryable())
            .clientError(category.isClientError())
            .build();
    }
    
    private ExceptionLevel determineLevel(ExceptionCategory category, Throwable cause) {
        if (cause instanceof OutOfMemoryError) return ExceptionLevel.FATAL;
        if (category.isClientError()) return ExceptionLevel.WARN;
        return ExceptionLevel.ERROR;
    }
}

---

2. ComponentUncaughtExceptionHandler Enhancement

Current Issues

• No Exception Context: Missing thread state, component info
• No Recovery Attempt: Immediately logs and reports without recovery
• Limited Diagnostics: No heap dump, thread dump for FATAL errors
• No Circuit Breaking: Doesn't prevent cascading failures

Solution: Graceful Exception Handling Pipeline

2.1 Enhanced Handler with Recovery Strategy

public class ComponentUncaughtExceptionHandler implements Thread.UncaughtExceptionHandler {
    
    private static final Logger LOGGER = LoggerFactory.getLogger(ComponentUncaughtExceptionHandler.class);
    public static final ComponentUncaughtExceptionHandler INSTANCE = new ComponentUncaughtExceptionHandler();
    
    private final ExceptionRecoveryOrchestrator recoveryOrchestrator;
    private final DiagnosticCollector diagnosticCollector;
    private final CircuitBreaker circuitBreaker;
    
    private ComponentUncaughtExceptionHandler() {
        this.recoveryOrchestrator = new ExceptionRecoveryOrchestrator();
        this.diagnosticCollector = new DiagnosticCollector();
        this.circuitBreaker = new CircuitBreaker(5, Duration.ofMinutes(1));
    }
    
    @Override
    public void uncaughtException(Thread thread, Throwable cause) {
        ThreadContext context = captureThreadContext(thread, cause);
        
        try {
            // Step 1: Classify exception severity
            ExceptionSeverity severity = classifyException(cause, context);
            
            // Step 2: Collect diagnostics if FATAL
            if (severity == ExceptionSeverity.FATAL) {
                diagnosticCollector.collectDiagnostics(thread, cause, context);
            }
            
            // Step 3: Check circuit breaker
            if (circuitBreaker.shouldTripFor(cause)) {
                LOGGER.error("Circuit breaker tripped for exception type: {}", 
                    cause.getClass().getName());
                initiateGracefulShutdown(cause, context);
                return;
            }
            
            // Step 4: Log with full context
            logException(severity, thread, cause, context);
            
            // Step 5: Report to metrics
            reportException(severity, cause, context);
            
            // Step 6: Attempt recovery
            RecoveryResult result = recoveryOrchestrator.attemptRecovery(
                thread, cause, severity, context);
            
            // Step 7: Handle recovery outcome
            if (!result.isSuccessful()) {
                handleRecoveryFailure(thread, cause, result, context);
            }
            
        } catch (Exception handlerException) {
            // Handler should never fail
            LOGGER.error("Exception handler itself failed", handlerException);
            forceShutdown(cause, handlerException);
        }
    }
    
    private ThreadContext captureThreadContext(Thread thread, Throwable cause) {
        return ThreadContext.builder()
            .threadId(thread.getId())
            .threadName(thread.getName())
            .threadGroup(thread.getThreadGroup().getName())
            .threadState(thread.getState())
            .stackTrace(thread.getStackTrace())
            .componentId(extractComponentId(thread))
            .timestamp(System.currentTimeMillis())
            .memorySnapshot(getMemorySnapshot())
            .activeTaskCount(getActiveTaskCount(thread))
            .build();
    }
    
    private ExceptionSeverity classifyException(Throwable cause, ThreadContext context) {
        if (cause instanceof OutOfMemoryError) return ExceptionSeverity.FATAL;
        if (cause instanceof GeaflowHeartbeatException) return ExceptionSeverity.CRITICAL;
        if (cause instanceof GeaflowInterruptedException) return ExceptionSeverity.WARNING;
        if (isRecoverableNetworkError(cause)) return ExceptionSeverity.WARNING;
        return ExceptionSeverity.ERROR;
    }
    
    private void logException(ExceptionSeverity severity, Thread thread, 
                             Throwable cause, ThreadContext context) {
        String logMessage = String.format(
            "[%s] Uncaught exception in thread '%s' (ID=%d, Component=%s)\n" +
            "Thread State: %s, Active Tasks: %d\n" +
            "Memory: %s\n" +
            "Root Cause: %s",
            severity,
            thread.getName(),
            thread.getId(),
            context.getComponentId(),
            context.getThreadState(),
            context.getActiveTaskCount(),
            context.getMemorySnapshot(),
            getRootCauseMessage(cause)
        );
        
        switch (severity) {
            case FATAL:
                LOGGER.error(logMessage, cause);
                break;
            case CRITICAL:
                LOGGER.error(logMessage, cause);
                break;
            case ERROR:
                LOGGER.error(logMessage, cause);
                break;
            case WARNING:
                LOGGER.warn(logMessage, cause);
                break;
        }
    }
    
    private void reportException(ExceptionSeverity severity, Throwable cause, ThreadContext context) {
        StatsCollectorFactory collectorFactory = StatsCollectorFactory.getInstance();
        if (collectorFactory != null) {
            ExceptionLevel level = mapSeverityToLevel(severity);
            collectorFactory.getExceptionCollector().reportException(level, cause);
            
            // Also report as event for critical/fatal
            if (severity.ordinal() <= ExceptionSeverity.CRITICAL.ordinal()) {
                collectorFactory.getEventCollector().reportEvent(
                    level, 
                    EventLabel.WORKER_PROCESS_EXITED,
                    buildEventMessage(context, cause)
                );
            }
        }
        
        // Add to supervisor queue
        ComponentExceptionSupervisor.getInstance().add(
            ComponentExceptionSupervisor.ExceptionElement.of(thread, cause, context));
    }
    
    private enum ExceptionSeverity {
        FATAL,     // OOM, system errors - immediate shutdown required
        CRITICAL,  // Heartbeat timeout, cluster split - needs urgent attention
        ERROR,     // Business logic errors - log and continue
        WARNING    // Transient errors - retry possible
    }
}

2.2 Recovery Orchestrator

public class ExceptionRecoveryOrchestrator {
    private final List<RecoveryStrategy> strategies;
    
    public ExceptionRecoveryOrchestrator() {
        strategies = Arrays.asList(
            new ThreadRestartStrategy(),
            new TaskResubmitStrategy(),
            new ResourceReleaseStrategy(),
            new GracefulDegradationStrategy()
        );
    }
    
    public RecoveryResult attemptRecovery(Thread thread, Throwable cause,
                                          ExceptionSeverity severity, ThreadContext context) {
        for (RecoveryStrategy strategy : strategies) {
            if (strategy.canHandle(cause, severity)) {
                try {
                    RecoveryAction action = strategy.createAction(thread, cause, context);
                    return action.execute();
                } catch (Exception e) {
                    LOGGER.warn("Recovery strategy {} failed", strategy.getName(), e);
                }
            }
        }
        return RecoveryResult.failed("No applicable recovery strategy");
    }
}

2.3 Diagnostic Collector

public class DiagnosticCollector {
    private final String diagnosticPath;
    
    public void collectDiagnostics(Thread thread, Throwable cause, ThreadContext context) {
        try {
            String timestamp = LocalDateTime.now().format(DateTimeFormatter.ISO_LOCAL_DATE_TIME);
            String diagnosticFile = String.format("%s/diagnostic_%s_%d.txt", 
                diagnosticPath, thread.getName(), timestamp);
            
            try (BufferedWriter writer = new BufferedWriter(new FileWriter(diagnosticFile))) {
                // Thread dump
                writer.write("=== THREAD DUMP ===\n");
                Map<Thread, StackTraceElement[]> allThreads = Thread.getAllStackTraces();
                for (Map.Entry<Thread, StackTraceElement[]> entry : allThreads.entrySet()) {
                    writer.write(formatThreadDump(entry.getKey(), entry.getValue()));
                }
                
                // Memory snapshot
                writer.write("\n=== MEMORY SNAPSHOT ===\n");
                writer.write(getDetailedMemoryInfo());
                
                // Exception stack trace
                writer.write("\n=== EXCEPTION STACK TRACE ===\n");
                writer.write(ExceptionUtils.getStackTrace(cause));
                
                // Context information
                writer.write("\n=== CONTEXT ===\n");
                writer.write(context.toString());
            }
            
            LOGGER.info("Diagnostics collected at: {}", diagnosticFile);
            
            // Optionally trigger heap dump for OOM
            if (cause instanceof OutOfMemoryError) {
                triggerHeapDump(timestamp);
            }
        } catch (Exception e) {
            LOGGER.error("Failed to collect diagnostics", e);
        }
    }
}

---

3. SliceOutputChannelHandler Enhancement

Current Issues

• Generic IOException: All errors wrapped in generic IOException
• No Error Classification: Network vs. business logic errors mixed
• No Retry Logic: Fails immediately without retry
• Limited Context: Missing channel state, buffer info
• No Metrics: No tracking of error patterns

Solution: Specific Exception Handling with Retry

3.1 Typed Exception Hierarchy

// Base exception
public class SliceTransferException extends IOException {
    private final ChannelId channelId;
    private final long sequenceNumber;
    private final ErrorCategory category;
    
    public enum ErrorCategory {
        NETWORK_TIMEOUT,
        CHANNEL_CLOSED,
        BUFFER_OVERFLOW,
        SERIALIZATION_ERROR,
        RESOURCE_EXHAUSTED,
        UNKNOWN
    }
}

// Specific exceptions
public class ChannelClosedException extends SliceTransferException {
    public ChannelClosedException(ChannelId channelId, String reason) {
        super(channelId, -1, ErrorCategory.CHANNEL_CLOSED, 
            String.format("Channel %s closed: %s", channelId, reason));
    }
}

public class BufferExhaustedException extends SliceTransferException {
    private final int currentBufferSize;
    private final int maxBufferSize;
}

public class NetworkTimeoutException extends SliceTransferException {
    private final long timeoutMs;
}

3.2 Enhanced Exception Handling

public class SliceOutputChannelHandler extends ChannelInboundHandlerAdapter {
    
    private static final Logger LOGGER = LoggerFactory.getLogger(SliceOutputChannelHandler.class);
    
    private final RetryPolicy retryPolicy;
    private final ExceptionMetricsCollector metricsCollector;
    private final Map<ChannelId, ChannelHealthTracker> healthTrackers;
    
    public SliceOutputChannelHandler() {
        this.retryPolicy = RetryPolicy.exponentialBackoff(3, 100, 1000);
        this.metricsCollector = new ExceptionMetricsCollector();
        this.healthTrackers = new ConcurrentHashMap<>();
    }
    
    @Override
    public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) throws Exception {
        ChannelContext channelContext = captureChannelContext(ctx);
        
        try {
            // Classify exception
            SliceTransferException typed = classifyException(cause, channelContext);
            
            // Log with context
            logChannelException(typed, channelContext);
            
            // Update metrics
            metricsCollector.recordException(typed);
            
            // Update health tracker
            updateHealthTracker(ctx.channel(), typed);
            
            // Determine handling strategy
            ExceptionHandlingStrategy strategy = determineStrategy(typed, channelContext);
            
            switch (strategy) {
                case RETRY:
                    handleRetriableException(ctx, typed, channelContext);
                    break;
                    
                case GRACEFUL_CLOSE:
                    handleGracefulClose(ctx, typed, channelContext);
                    break;
                    
                case IMMEDIATE_CLOSE:
                    handleImmediateClose(ctx, typed, channelContext);
                    break;
                    
                case IGNORE:
                    // Already logged, do nothing
                    break;
            }
            
        } catch (Exception handlerException) {
            LOGGER.error("Exception handler failed", handlerException);
            handleException(ctx.channel(), cause); // Fallback to original
        }
    }
    
    private SliceTransferException classifyException(Throwable cause, ChannelContext context) {
        // Network errors
        if (cause instanceof SocketTimeoutException) {
            return new NetworkTimeoutException(
                context.getChannelId(), 
                ((SocketTimeoutException) cause).bytesTransferred,
                context.getTimeout()
            );
        }
        
        if (cause instanceof ClosedChannelException) {
            return new ChannelClosedException(
                context.getChannelId(),
                "Channel closed during operation"
            );
        }
        
        // Buffer errors
        if (cause instanceof OutOfMemoryError || cause.getMessage().contains("buffer")) {
            return new BufferExhaustedException(
                context.getChannelId(),
                context.getCurrentBufferSize(),
                context.getMaxBufferSize()
            );
        }
        
        // Serialization errors
        if (cause instanceof IOException && context.isSerializationError(cause)) {
            return new SerializationException(
                context.getChannelId(),
                context.getCurrentSequence(),
                cause
            );
        }
        
        // Unknown - wrap with context
        return new SliceTransferException(
            context.getChannelId(),
            context.getCurrentSequence(),
            SliceTransferException.ErrorCategory.UNKNOWN,
            cause.getMessage(),
            cause
        );
    }
    
    private void logChannelException(SliceTransferException exception, ChannelContext context) {
        String logMessage = String.format(
            "[%s] Slice transfer error on channel %s\n" +
            "Category: %s, Sequence: %d\n" +
            "Channel State: %s, Active Readers: %d\n" +
            "Available Buffers: %d/%d\n" +
            "Error: %s",
            exception.getCategory(),
            context.getChannelId(),
            exception.getCategory(),
            exception.getSequenceNumber(),
            context.getChannelState(),
            context.getActiveReaderCount(),
            context.getAvailableBuffers(),
            context.getTotalBuffers(),
            exception.getMessage()
        );
        
        if (exception.getCategory() == SliceTransferException.ErrorCategory.NETWORK_TIMEOUT) {
            LOGGER.warn(logMessage, exception);
        } else {
            LOGGER.error(logMessage, exception);
        }
    }
    
    private void handleRetriableException(ChannelHandlerContext ctx, 
                                          SliceTransferException exception,
                                          ChannelContext channelContext) {
        RetryContext retryCtx = channelContext.getRetryContext();
        
        if (retryPolicy.shouldRetry(retryCtx)) {
            long delayMs = retryPolicy.nextRetryDelay(retryCtx);
            
            LOGGER.info("Scheduling retry {} of {} in {}ms for channel {}", 
                retryCtx.getAttempt() + 1, 
                retryPolicy.getMaxAttempts(),
                delayMs,
                channelContext.getChannelId());
            
            ctx.executor().schedule(() -> {
                try {
                    retryTransfer(ctx, channelContext);
                } catch (Exception e) {
                    LOGGER.error("Retry failed for channel {}", 
                        channelContext.getChannelId(), e);
                    handleImmediateClose(ctx, exception, channelContext);
                }
            }, delayMs, TimeUnit.MILLISECONDS);
        } else {
            LOGGER.error("Retry limit exceeded for channel {}, closing", 
                channelContext.getChannelId());
            handleImmediateClose(ctx, exception, channelContext);
        }
    }
    
    private void handleGracefulClose(ChannelHandlerContext ctx,
                                      SliceTransferException exception,
                                      ChannelContext channelContext) {
        LOGGER.info("Initiating graceful close for channel {}", channelContext.getChannelId());
        
        try {
            // Flush pending writes
            ctx.flush();
            
            // Send error response
            ErrorResponse errorResponse = new ErrorResponse(exception);
            ctx.writeAndFlush(errorResponse).addListener(future -> {
                if (future.isSuccess()) {
                    releaseResources(ctx, channelContext);
                    ctx.close();
                } else {
                    LOGGER.warn("Failed to send error response, force closing");
                    handleImmediateClose(ctx, exception, channelContext);
                }
            });
        } catch (Exception e) {
            LOGGER.error("Graceful close failed", e);
            handleImmediateClose(ctx, exception, channelContext);
        }
    }
    
    private void handleImmediateClose(ChannelHandlerContext ctx,
                                       SliceTransferException exception,
                                       ChannelContext channelContext) {
        LOGGER.error("Force closing channel {} due to: {}", 
            channelContext.getChannelId(), exception.getMessage());
        
        fatalError = true;
        
        try {
            releaseAllResources();
        } catch (IOException e) {
            LOGGER.error("Failed to release resources", e);
        }
        
        if (ctx.channel().isActive()) {
            ctx.channel().close();
        }
    }
    
    private ChannelContext captureChannelContext(ChannelHandlerContext ctx) {
        return ChannelContext.builder()
            .channel(ctx.channel())
            .channelId(extractChannelId(ctx))
            .channelState(ctx.channel().isActive() ? "ACTIVE" : "INACTIVE")
            .currentSequence(getCurrentSequence())
            .activeReaderCount(allReaders.size())
            .availableBuffers(availableReaders.size())
            .totalBuffers(allReaders.size())
            .timeout(getConfiguredTimeout())
            .build();
    }
}

3.3 Retry Policy

public class RetryPolicy {
    private final int maxAttempts;
    private final long initialDelayMs;
    private final long maxDelayMs;
    private final double multiplier;
    
    public static RetryPolicy exponentialBackoff(int maxAttempts, 
                                                 long initialDelayMs, 
                                                 long maxDelayMs) {
        return new RetryPolicy(maxAttempts, initialDelayMs, maxDelayMs, 2.0);
    }
    
    public boolean shouldRetry(RetryContext context) {
        return context.getAttempt() < maxAttempts;
    }
    
    public long nextRetryDelay(RetryContext context) {
        long delay = (long) (initialDelayMs * Math.pow(multiplier, context.getAttempt()));
        return Math.min(delay, maxDelayMs);
    }
}

[yazong]

#720