refactor: unalignworker implement align

## Background

```mermaid
graph TD
A["Scheduler sends event"] --> B["process(fetchCount, isAligned=false)"]
B --> C{executorService is null?}
C -->|Yes| D["startTask()"]
C -->|No| E["Continue processing"]

D --> F["Create ExecutorService"]
F --> G["Start WorkerTask thread"]

G --> H["WorkerTask.run()"]
H --> I["while(running) loop"]
I --> J["unalignedProcess()"]

J --> K["inputReader.poll() get message"]
K --> L{Message type check}

L -->|Data message| M["processMessage(windowId, message)"]
L -->|Barrier message| N["processBarrier(windowId, totalCount)"]
L -->|No message| O["Continue polling"]

M --> P["processMessageEvent() process data"]
P --> Q["Update windowCount"]
Q --> I

N --> R["Verify processing count"]
R --> S["Get windowId from processingWindowIdQueue"]
S --> T["Call finish() to complete processing"]
T --> U["Initialize next window"]
U --> I

V["interrupt()"] --> W["Shutdown ExecutorService"]
X["close()"] --> Y["Clean up resources and queues"]
```

## Core Mechanism Explanation

### 1. Unaligned Processing Mode
`AbstractUnAlignedWorker` extends `AbstractComputeWorker` and specializes in handling unaligned data streams. Unlike aligned mode, it uses independent threads to process data asynchronously without waiting for all upstream data to arrive.

### 2. Asynchronous Execution Mechanism
When `isAligned=false`, the system starts a dedicated `ExecutorService` and `WorkerTask` thread. This thread continuously runs the `unalignedProcess()` method, polling the input queue for messages.

### 3. Message Processing Flow
- **Data message processing**: Directly calls `processMessageEvent()` to process data
- **Barrier message processing**: Verifies processing count, retrieves window ID from `processingWindowIdQueue`, and completes current window processing

### 4. Window Management
Uses `processingWindowIdQueue` to manage window IDs being processed, ensuring correct window processing order. In `finishBarrier()`, it validates that the current window ID must be within a reasonable range.

## Project Status Analysis

The alignment processing of LoadGraphProcessEvent in the current system has the following status:

1. **Current Implementation Mechanism**: A TODO comment in AbstractUnAlignedWorker indicates that LoadGraphProcessEvent needs to be aligned

2. **Alignment Judgment Logic**: In the execute method of AbstractIterationComputeCommand, the instanceof method is used to check LoadGraphProcessEvent to force alignment

3. **Processing Flow**: The process method of AbstractUnAlignedWorker determines whether to call alignedProcess or start asynchronous processing based on the isAligned parameter

## Development Plan

### Phase 1: Improve the alignment processing mechanism of LoadGraphProcessEvent

**Goal**: Ensure that LoadGraphProcessEvent can correctly trigger alignment processing in all worker types

**Specific Tasks**:

1. **Improve the alignment processing in AbstractUnAlignedWorker**
- Remove the TODO comment on line 51 and implement the complete alignment processing logic of LoadGraphProcessEvent
- Ensure that the alignedProcess method can correctly handle the special requirements of LoadGraphProcessEvent
- Add the necessary synchronization mechanism to ensure the consistency of graph data loading

2. **Optimize the LoadGraphProcessEvent class itself**
- Add the alignment processing identification property in LoadGraphProcessEvent
- Rewrite the relevant methods to ensure the correct execution of the alignment processing

3. **Enhance the event construction in SchedulerEventBuilder**
- Improve the creation logic of LoadGraphProcessEvent in the buildInitIteration method
- Add the necessary configuration parameters to support the customization of the alignment processing

### Phase 2: Unified alignment processing framework

**Goal**: Establish a unified alignment processing framework to support the alignment requirements of different event types

**Specific tasks**:

1. **Abstract alignment processing interface**
- Create the AlignmentRequired interface to identify events that require alignment processing
- LoadGraphProcessEvent implements this interface

2. **Enhance AbstractIterationComputeCommand**
- Optimize the alignment judgment logic in the execute method

- Support more flexible alignment strategy configuration

3. **Improve the Worker class hierarchy**

- Provide a unified alignment processing basic method in AbstractWorker

- Ensure the implementation consistency of AlignedComputeWorker and UnAlignedComputeWorker

### Phase 3: Performance optimization and testing

**Goal**: Optimize alignment processing performance and ensure system stability

## Post-Refactoring AbstractUnAlignedWorker Flow Diagram

```mermaid
flowchart TD
A["Start Event Processing"] --> B{"Check isAligned Parameter"}
B -->|true| C["Execute alignedProcess"]
B -->|false| D{"Check if executorService is null"}
D -->|null| E["Start Async Task"]
D -->|not null| F["Continue Async Processing"]

E --> G["Create ExecutorService"]
G --> H["Start WorkerTask Thread"]

C --> I["Call super.process Synchronously"]
I --> J["Processing Complete"]

H --> K["Loop Execute unalignedProcess"]
K --> L["Poll Messages from inputReader"]
L --> M{"Message Type Check"}
M -->|"Has Message"| N["Process Message Event"]
M -->|"No Message but Window Count"| O["Process Barrier Event"]
M -->|"No Message"| K

N --> P["Call processMessageEvent"]
O --> Q["Call finishBarrier"]
P --> K
Q --> K

%% New alignment processing logic
subgraph "New Alignment Processing Framework"
R["AlignmentUtil.requiresAlignment Check"]
S["AlignmentManager Coordination"]
T["shouldUseAlignedProcessing Decision"]
end

B -.->|"New Check"| R
R --> T
T --> B

style R fill:#ffeb3b
style S fill:#ffeb3b
style T fill:#ffeb3b
```

## Post-Refactoring Interaction Sequence Diagram

```mermaid
sequenceDiagram
participant Scheduler as Scheduler
participant Command as AbstractIterationComputeCommand
participant Worker as AbstractUnAlignedWorker
participant AlignmentUtil as AlignmentUtil
participant AlignmentManager as AlignmentManager
participant ExecutorService as ExecutorService
participant InputReader as InputReader

Note over Scheduler,InputReader: Post-Refactoring Alignment Processing Flow

Scheduler->>Command: execute(taskContext)
Command->>Command: determineAlignmentRequirement()

%% New alignment check logic
rect rgb(255, 235, 59, 0.3)
Note over Command,AlignmentUtil: NEW: Alignment Requirement Check
Command->>AlignmentUtil: requiresAlignment(event)
AlignmentUtil-->>Command: boolean result
Command->>Worker: shouldUseAlignedProcessing(event, isAligned)
Worker-->>Command: enhanced alignment decision
end

Command->>Worker: process(fetchCount, requiresAlignment)

alt requiresAlignment = true
Worker->>Worker: alignedProcess(fetchCount)
Worker->>Worker: super.process(fetchCount, true)
Note over Worker: Synchronous Processing Mode
else requiresAlignment = false
Worker->>Worker: Check executorService
alt executorService == null
Worker->>ExecutorService: Create and Start
ExecutorService->>Worker: Start WorkerTask Thread
end

loop Async Processing Loop
Worker->>InputReader: poll(timeout)
InputReader-->>Worker: InputMessage
alt Has Message
Worker->>Worker: processMessage(windowId, message)
else Has Barrier
Worker->>Worker: processBarrier(windowId, totalCount)

%% New alignment management
rect rgb(255, 235, 59, 0.3)
Note over Worker,AlignmentManager: NEW: Alignment Manager Processing
Worker->>AlignmentManager: Optional Alignment Coordination
end
end
end
end

%% New resource cleanup
rect rgb(255, 235, 59, 0.3)
Note over Worker,AlignmentManager: NEW: Enhanced Resource Cleanup
Worker->>AlignmentManager: shutdown()
AlignmentManager-->>Worker: Cleanup Complete
end
```

## Before vs After Comparison Analysis

### Pre-Refactoring AbstractUnAlignedWorker [1](#6-0)

The pre-refactoring implementation had the following issues:
1. **Hard-coded alignment judgment**: Could only identify LoadGraphProcessEvent alignment needs through TODO comments
2. **Lack of extensibility**: Unable to support alignment requirements for other event types
3. **Inflexible configuration**: No unified alignment configuration management

### Major Changes After Refactoring

#### 1. New Alignment Processing Interface Framework
- **AlignmentRequired Interface**: Standardized alignment requirement declaration
- **AlignmentUtil Utility Class**: Unified alignment logic judgment
- **AlignmentManager**: Coordinates multi-worker alignment processing

#### 2. Enhanced Alignment Decision Logic
```java
// Before: Hard-coded judgment
// TODO Currently LoadGraphProcessEvent need align processing.
if (isAligned) {
alignedProcess(fetchCount);
}

// After: Intelligent judgment
public boolean shouldUseAlignedProcessing(IEvent event, boolean isAligned) {
if (AlignmentUtil.requiresAlignment(event)) {
return true;
}
return isAligned;
}
```

#### 3. Improved Resource Management
After refactoring, AlignmentManager cleanup was added to the `interrupt()` and `close()` methods:
```java
if (alignmentManager != null) {
alignmentManager.shutdown();
}
```

### Basic data structure

#### Core Components

##### 1. AlignmentRequired Interface

Identifies the event interface that requires alignment processing and provides the following configuration options:

- `requiresStrictAlignment()`: Whether strict alignment is required (default true)

- `getAlignmentTimeoutMs()`: Alignment timeout (default 30 seconds)

- `getAlignmentStrategy()`: Alignment strategy (STRICT/RELAXED/CUSTOM)

##### 2. AlignmentConfig Configuration Class

Global configuration of alignment processing:

```java
AlignmentConfig config = AlignmentConfig.builder()
.strategy(AlignmentStrategy.STRICT)
.timeoutMs(30000L)
.enableRetry(true)
.maxRetryAttempts(3)
.retryDelayMs(1000L)
.build();
```

##### 3. AlignmentManager

Responsible for coordinating alignment across multiple workers:

- Manages alignment barriers
- Handles timeouts and retry logic
- Provides a graceful shutdown mechanism

How to use

1. Create a LoadGraphProcessEvent
```
// Uses the default alignment configuration
LoadGraphProcessEvent event = new LoadGraphProcessEvent(
schedulerId, workerId, cycleId, windowId, fetchWindowId, fetchCount);

// Uses a custom alignment configuration
LoadGraphProcessEvent customEvent = new LoadGraphProcessEvent(
schedulerId, workerId, cycleId, windowId, fetchWindowId, fetchCount,
AlignmentStrategy.RELAXED, 60000L);
```
2. Handles alignment in the worker

AbstractUnAlignedWorker automatically detects the LoadGraphProcessEvent and enables alignment:
```
// In AbstractIterationComputeCommand.execute()
boolean requiresAlignment = determineAlignmentRequirement(abstractWorker);
abstractWorker.process(fetchCount, requiresAlignment);
```
3. Configure the alignment strategy

- STRICT: Strict alignment, all workers must be synchronized
- RELAXED: Relaxed alignment, best-effort synchronization
- CUSTOM: Custom alignment logic

### 1. Create an alignment processing interface

create an event interface that identifies the events that need to be aligned:
```java

package org.apache.geaflow.runtime.core.protocol;

/**
* Interface to mark events that require aligned processing.
* Events implementing this interface will be forced to use aligned processing
* regardless of the worker type to ensure data consistency.
*/
public interface AlignmentRequired {

/**
* Returns whether this event requires strict alignment.
* @return true if strict alignment is required, false otherwise
*/
default boolean requiresStrictAlignment() {
return true;
}

/**
* Returns the alignment timeout in milliseconds.
* @return timeout value, -1 for no timeout
*/
default long getAlignmentTimeoutMs() {
return -1L;
}

/**
* Returns the alignment strategy for this event.
* @return alignment strategy
*/
default AlignmentStrategy getAlignmentStrategy() {
return AlignmentStrategy.STRICT;
}

/**
* Alignment strategies for different event types.
*/
enum AlignmentStrategy {
/** Strict alignment - all workers must be synchronized */
STRICT,
/** Relaxed alignment - best effort synchronization */
RELAXED,
/** Custom alignment - event-specific logic */
CUSTOM
}
}
```

### 2. Optimize the LoadGraphProcessEvent class

Modify the existing LoadGraphProcessEvent class to implement the alignment processing interface:

```java

package org.apache.geaflow.runtime.core.protocol;

import org.apache.geaflow.cluster.fetcher.CloseFetchRequest;
import org.apache.geaflow.cluster.protocol.EventType;
import org.apache.geaflow.cluster.task.ITaskContext;
import org.apache.geaflow.runtime.core.worker.context.WorkerContext;

public class LoadGraphProcessEvent extends AbstractIterationComputeCommand implements AlignmentRequired {

private static final long DEFAULT_ALIGNMENT_TIMEOUT_MS = 30000L; // 30 seconds
private final AlignmentStrategy alignmentStrategy;
private final long alignmentTimeoutMs;

public LoadGraphProcessEvent(long schedulerId, int workerId, int cycleId, long windowId, long fetchWindowId, long fetchCount) {
this(schedulerId, workerId, cycleId, windowId, fetchWindowId, fetchCount, AlignmentStrategy.STRICT, DEFAULT_ALIGNMENT_TIMEOUT_MS);
}

public LoadGraphProcessEvent(long schedulerId, int workerId, int cycleId, long windowId,
long fetchWindowId, long fetchCount, AlignmentStrategy alignmentStrategy,
long alignmentTimeoutMs) {
super(schedulerId, workerId, cycleId, windowId, fetchWindowId, fetchCount);
this.alignmentStrategy = alignmentStrategy;
this.alignmentTimeoutMs = alignmentTimeoutMs;
}

@Override
public void execute(ITaskContext taskContext) {
super.execute(taskContext);
this.fetcherRunner.add(new CloseFetchRequest(((WorkerContext) this.context).getTaskId()));
}

@Override
public EventType getEventType() {
return EventType.PRE_GRAPH_PROCESS;
}

@Override
public boolean requiresStrictAlignment() {
return alignmentStrategy == AlignmentStrategy.STRICT;
}

@Override
public long getAlignmentTimeoutMs() {
return alignmentTimeoutMs;
}

@Override
public AlignmentStrategy getAlignmentStrategy() {
return alignmentStrategy;
}

@Override
public String toString() {
return "LoadGraphProcessEvent{"
+ "schedulerId=" + schedulerId
+ ", workerId=" + workerId
+ ", cycleId=" + cycleId
+ ", windowId=" + windowId
+ ", fetchWindowId=" + fetchWindowId
+ ", alignmentStrategy=" + alignmentStrategy
+ ", alignmentTimeoutMs=" + alignmentTimeoutMs
+ '}';
}
}
```

### 3. Create an alignment processing tool class

Create a utility class to handle the alignment logic:

```java

package org.apache.geaflow.runtime.core.worker.alignment;

import org.apache.geaflow.runtime.core.protocol.AlignmentRequired;
import org.apache.geaflow.runtime.core.protocol.IEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* Utility class for handling alignment logic in workers.
*/
public class AlignmentUtil {

private static final Logger LOGGER = LoggerFactory.getLogger(AlignmentUtil.class);

/**
* Determines if an event requires aligned processing.
*
* @param event the event to check
* @return true if the event requires alignment, false otherwise
*/
public static boolean requiresAlignment(IEvent event) {
if (event instanceof AlignmentRequired) {
AlignmentRequired alignmentRequired = (AlignmentRequired) event;
boolean requires = alignmentRequired.requiresStrictAlignment();
LOGGER.debug("Event {} requires alignment: {}", event.getClass().getSimpleName(), requires);
return requires;
}
return false;
}

/**
* Gets the alignment timeout for an event.
*
* @param event the event to check
* @return timeout in milliseconds, or -1 if no timeout
*/
public static long getAlignmentTimeout(IEvent event) {
if (event instanceof AlignmentRequired) {
return ((AlignmentRequired) event).getAlignmentTimeoutMs();
}
return -1L;
}

/**
* Gets the alignment strategy for an event.
*
* @param event the event to check
* @return the alignment strategy
*/
public static AlignmentRequired.AlignmentStrategy getAlignmentStrategy(IEvent event) {
if (event instanceof AlignmentRequired) {
return ((AlignmentRequired) event).getAlignmentStrategy();
}
return AlignmentRequired.AlignmentStrategy.STRICT;
}

/**
* Validates alignment configuration for an event.
*
* @param event the event to validate
* @throws IllegalArgumentException if configuration is invalid
*/
public static void validateAlignmentConfig(IEvent event) {
if (event instanceof AlignmentRequired) {
AlignmentRequired alignmentRequired = (AlignmentRequired) event;
long timeout = alignmentRequired.getAlignmentTimeoutMs();
if (timeout > 0 && timeout < 1000) {
throw new IllegalArgumentException("Alignment timeout must be at least 1000ms or -1 for no timeout");
}
}
}
}
```

### 4. Optimizing the AbstractUnAlignedWorker class

Modify AbstractUnAlignedWorker to support the new alignment processing logic:
```java

package org.apache.geaflow.runtime.core.worker;

import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import org.apache.geaflow.cluster.task.ITaskContext;
import org.apache.geaflow.common.exception.GeaflowRuntimeException;
import org.apache.geaflow.common.utils.Executors;
import org.apache.geaflow.runtime.core.protocol.IEvent;
import org.apache.geaflow.runtime.core.worker.alignment.AlignmentUtil;
import org.apache.geaflow.runtime.core.worker.context.WorkerContext;
import org.apache.geaflow.shuffle.message.InputMessage;
import org.apache.geaflow.shuffle.message.PipelineMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractUnAlignedWorker extends AbstractComputeWorker {

private static final Logger LOGGER = LoggerFactory.getLogger(AbstractUnAlignedWorker.class);

private static final String WORKER_FORMAT = "geaflow-asp-worker-";
private static final int DEFAULT_TIMEOUT_MS = 100;

protected ExecutorService executorService;
protected BlockingQueue processingWindowIdQueue;

public AbstractUnAlignedWorker() {
super();
this.processingWindowIdQueue = new LinkedBlockingDeque<>();
}

@Override
public void process(long fetchCount, boolean isAligned) {
// Enhanced alignment processing logic for LoadGraphProcessEvent and other alignment-required events
if (isAligned) {
LOGGER.debug("Processing with aligned mode for taskId {}", context.getTaskId());
alignedProcess(fetchCount);
} else {
if (executorService == null) {
LOGGER.info("taskId {} unaligned worker has been shutdown, start...", context.getTaskId());
startTask();
}
}
}

/**
* Enhanced method to determine if an event requires aligned processing.
* This method checks both the isAligned parameter and the event's alignment requirements.
*/
public boolean shouldUseAlignedProcessing(IEvent event, boolean isAligned) {
// Check if event explicitly requires alignment
if (AlignmentUtil.requiresAlignment(event)) {
LOGGER.debug("Event {} requires aligned processing", event.getClass().getSimpleName());
return true;
}

// Fall back to the original isAligned parameter
return isAligned;
}

private void startTask() {
long start = System.currentTimeMillis();
this.executorService = Executors.getExecutorService(1, WORKER_FORMAT + context.getTaskId() + "-%d",
ComponentUncaughtExceptionHandler.INSTANCE);
executorService.execute(new WorkerTask());
LOGGER.info("taskId {} start task cost {}ms", context != null ? context.getTaskId() : "null", System.currentTimeMillis() - start);
}

public void alignedProcess(long fetchCount) {
LOGGER.debug("Executing aligned process for taskId {} with fetchCount {}", context.getTaskId(), fetchCount);
super.process(fetchCount, true);
}

public void unalignedProcess() {
try {
InputMessage input = inputReader.poll(DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
if (input != null) {
long windowId = input.getWindowId();
if (input.getMessage() != null) {
PipelineMessage message = input.getMessage();
processMessage(windowId, message);
} else {
long totalCount = input.getWindowCount();
processBarrier(windowId, totalCount);
}
}
} catch (Throwable t) {
if (running) {
LOGGER.error(t.getMessage(), t);
throw new GeaflowRuntimeException(t);
} else {
LOGGER.warn("service closed {}", t.getMessage());
}
}

if (!running) {
LOGGER.info("{} worker terminated",
context == null ? "null" : context.getTaskId());
}
}

// ... rest of the existing methods remain unchanged
}
```

### 5. Create an alignment processing configuration class

```java
ublic class AlignmentConfig {

private static final long DEFAULT_TIMEOUT_MS = 30000L;
private static final AlignmentStrategy DEFAULT_STRATEGY = AlignmentStrategy.STRICT;

private final AlignmentStrategy strategy;
private final long timeoutMs;
private final boolean enableRetry;
private final int maxRetryAttempts;
private final long retryDelayMs;

public AlignmentConfig() {
this(DEFAULT_STRATEGY, DEFAULT_TIMEOUT_MS, true, 3, 1000L);
}

public AlignmentConfig(AlignmentStrategy strategy, long timeoutMs, boolean enableRetry,
int maxRetryAttempts, long retryDelayMs) {
this.strategy = strategy;
this.timeoutMs = timeoutMs;
this.enableRetry = enableRetry;
this.maxRetryAttempts = maxRetryAttempts;
this.retryDelayMs = retryDelayMs;
}

public AlignmentStrategy getStrategy() {
return strategy;
}

public long getTimeoutMs() {
return timeoutMs;
}

public boolean isEnableRetry() {
return enableRetry;
}

public int getMaxRetryAttempts() {
return maxRetryAttempts;
}

public long getRetryDelayMs() {
return retryDelayMs;
}

public static Builder builder() {
return new Builder();
}

public static class Builder {
private AlignmentStrategy strategy = DEFAULT_STRATEGY;
private long timeoutMs = DEFAULT_TIMEOUT_MS;
private boolean enableRetry = true;
private int maxRetryAttempts = 3;
private long retryDelayMs = 1000L;

public Builder strategy(AlignmentStrategy strategy) {
this.strategy = strategy;
return this;
}

public Builder timeoutMs(long timeoutMs) {
this.timeoutMs = timeoutMs;
return this;
}

public Builder enableRetry(boolean enableRetry) {
this.enableRetry = enableRetry;
return this;
}

public Builder maxRetryAttempts(int maxRetryAttempts) {
this.maxRetryAttempts = maxRetryAttempts;
return this;
}

public Builder retryDelayMs(long retryDelayMs) {
this.retryDelayMs = retryDelayMs;
return this;
}

public AlignmentConfig build() {
return new AlignmentConfig(strategy, timeoutMs, enableRetry, maxRetryAttempts, retryDelayMs);
}
}
}
```
6.Creating an Alignment Manager

```
package org.apache.geaflow.runtime.core.worker.alignment;

import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import org.apache.geaflow.common.exception.GeaflowRuntimeException;
import org.apache.geaflow.runtime.core.protocol.AlignmentRequired;
import org.apache.geaflow.runtime.core.protocol.IEvent;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
* Manager for handling alignment processing across workers.
*/
public class AlignmentManager {

private static final Logger LOGGER = LoggerFactory.getLogger(AlignmentManager.class);

private final ConcurrentHashMap alignmentBarriers;
private final AlignmentConfig config;
private final AtomicBoolean isShutdown;

public AlignmentManager(AlignmentConfig config) {
this.config = config;
this.alignmentBarriers = new ConcurrentHashMap<>();
this.isShutdown = new AtomicBoolean(false);
}

/**
* Waits for alignment barrier for the given event.
*
* @param event the event requiring alignment
* @param workerId the worker ID
* @param totalWorkers total number of workers
* @return true if alignment succeeded, false if timeout or interrupted
*/
public boolean waitForAlignment(IEvent event, int workerId, int totalWorkers) {
if (isShutdown.get()) {
return false;
}

String barrierKey = generateBarrierKey(event);
AlignmentBarrier barrier = alignmentBarriers.computeIfAbsent(barrierKey,
k -> new AlignmentBarrier(totalWorkers, AlignmentUtil.getAlignmentTimeout(event)));

try {
boolean success = barrier.await(workerId);
if (success) {
LOGGER.debug("Worker {} successfully aligned for event {}", workerId, event.getClass().getSimpleName());
} else {
LOGGER.warn("Worker {} alignment timeout for event {}", workerId, event.getClass().getSimpleName());
}
return success;
} finally {
// Clean up completed barriers
if (barrier.isCompleted()) {
alignmentBarriers.remove(barrierKey);
}
}
}

/**
* Signals that a worker has reached the alignment point.
*
* @param event the event requiring alignment
* @param workerId the worker ID
*/
public void signalAlignment(IEvent event, int workerId) {
if (isShutdown.get()) {
return;
}

String barrierKey = generateBarrierKey(event);
AlignmentBarrier barrier = alignmentBarriers.get(barrierKey);
if (barrier != null) {
barrier.signal(workerId);
LOGGER.debug("Worker {} signaled alignment for event {}", workerId, event.getClass().getSimpleName());
}
}

/**
* Shuts down the alignment manager and releases all resources.
*/
public void shutdown() {
isShutdown.set(true);
alignmentBarriers.values().forEach(AlignmentBarrier::forceComplete);
alignmentBarriers.clear();
LOGGER.info("AlignmentManager shutdown completed");
}

private String generateBarrierKey(IEvent event) {
return event.getClass().getSimpleName() + "_" + System.identityHashCode(event);
}

/**
* Internal class representing an alignment barrier for a specific event.
*/
private static class AlignmentBarrier {
private final CountDownLatch latch;
private final long timeoutMs;
private final AtomicBoolean completed;

public AlignmentBarrier(int workerCount, long timeoutMs) {
this.latch = new CountDownLatch(workerCount);
this.timeoutMs = timeoutMs > 0 ? timeoutMs : 30000L; // Default 30s timeout
this.completed = new AtomicBoolean(false);
}

public boolean await(int workerId) {
try {
boolean success = timeoutMs > 0 ?
latch.await(timeoutMs, TimeUnit.MILLISECONDS) :
(latch.await() || true); // await() returns void, so we return true
completed.set(true);
return success;
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
completed.set(true);
return false;
}
}

public void signal(int workerId) {
latch.countDown();
}

public boolean isCompleted() {
return completed.get();
}

public void forceComplete() {
while (latch.getCount() > 0) {
latch.countDown();
}
completed.set(true);
}
}
```

7.Optimizing the SchedulerEventBuilder class

```
package org.apache.geaflow.runtime.core.scheduler;

// ... existing imports ...
import org.apache.geaflow.runtime.core.protocol.AlignmentRequired.AlignmentStrategy;
import org.apache.geaflow.runtime.core.worker.alignment.AlignmentConfig;

public class SchedulerEventBuilder {

// ... existing fields ...
private final AlignmentConfig alignmentConfig;

public SchedulerEventBuilder(/* existing parameters */) {
// ... existing initialization ...
this.alignmentConfig = AlignmentConfig.builder()
.strategy(AlignmentStrategy.STRICT)
.timeoutMs(30000L)
.enableRetry(true)
.maxRetryAttempts(3)
.build();
}

// Enhanced buildInitIteration method with alignment support
private ExecutableEventIterator buildInitIteration(long iterationId) {
ExecutableEventIterator iterator = new ExecutableEventIterator();
for (ExecutionTask task : this.cycle.getTasks()) {
if (ExecutionTaskUtils.isCycleHead(task)) {
int workerId = task.getWorkerInfo().getWorkerIndex();

// Create LoadGraphProcessEvent with alignment configuration
IEvent loadGraph = createLoadGraphProcessEvent(workerId, iterationId);

// Init iteration.
IoDescriptor ioDescriptor = IoDescriptorBuilder.buildIterationIoDescriptor(
task, this.cycle, this.resultManager, OutputType.LOOP);
InitIterationEvent iterationInit = new InitIterationEvent(
this.schedulerId,
workerId,
this.cycle.getCycleId(),
iterationId,
this.cycle.getPipelineId(),
this.cycle.getPipelineName(),
ioDescriptor);
IEvent execute = new ExecuteFirstIterationEvent(this.schedulerId, workerId, this.cycle.getCycleId(), iterationId);
ComposeEvent composeEvent = new ComposeEvent(workerId,
Arrays.asList(loadGraph, iterationInit, execute));
iterator.addEvent(task.getWorkerInfo(), task, composeEvent);
}
}
return iterator;
}

/**
* Creates LoadGraphProcessEvent with proper alignment configuration.
*/
private LoadGraphProcessEvent createLoadGraphProcessEvent(int workerId, long iterationId) {
return new LoadGraphProcessEvent(
this.schedulerId,
workerId,
cycle.getCycleId(),
iterationId,
context.getInitialIterationId(),
COMPUTE_FETCH_COUNT,
alignmentConfig.getStrategy(),
alignmentConfig.getTimeoutMs()
);
}

/**
* Gets the alignment configuration for this scheduler.
*/
public AlignmentConfig getAlignmentConfig() {
return alignmentConfig;
}

// ... rest of existing methods remain unchanged
}
```

8.Improve the complete implementation of AbstractUnAlignedWorker

```
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingDeque;
import java.util.concurrent.TimeUnit;
import org.apache.geaflow.cluster.exception.ComponentUncaughtExceptionHandler;
import org.apache.geaflow.cluster.protocol.InputMessage;
import org.apache.geaflow.common.exception.GeaflowRuntimeException;
import org.apache.geaflow.common.thread.Executors;
import org.apache.geaflow.common.utils.ExecutorUtil;
import org.apache.geaflow.runtime.core.protocol.IEvent;
import org.apache.geaflow.runtime.core.worker.alignment.AlignmentUtil;
import org.apache.geaflow.runtime.core.worker.alignment.AlignmentManager;
import org.apache.geaflow.runtime.core.worker.alignment.AlignmentConfig;
import org.apache.geaflow.shuffle.message.PipelineMessage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class AbstractUnAlignedWorker extends AbstractComputeWorker {

private static final Logger LOGGER = LoggerFactory.getLogger(AbstractUnAlignedWorker.class);

private static final String WORKER_FORMAT = "geaflow-asp-worker-";
private static final int DEFAULT_TIMEOUT_MS = 100;

protected ExecutorService executorService;
protected BlockingQueue processingWindowIdQueue;
protected AlignmentManager alignmentManager;

public AbstractUnAlignedWorker() {
super();
this.processingWindowIdQueue = new LinkedBlockingDeque<>();
this.alignmentManager = new AlignmentManager(new AlignmentConfig());
}

@Override
public void process(long fetchCount, boolean isAligned) {
// Enhanced alignment processing logic for LoadGraphProcessEvent and other alignment-required events
if (isAligned) {
LOGGER.debug("Processing with aligned mode for taskId {}", context.getTaskId());
alignedProcess(fetchCount);
} else {
if (executorService == null) {
LOGGER.info("taskId {} unaligned worker has been shutdown, start...", context.getTaskId());
startTask();
}
}
}

/**
* Enhanced method to determine if an event requires aligned processing.
* This method checks both the isAligned parameter and the event's alignment requirements.
*/
public boolean shouldUseAlignedProcessing(IEvent event, boolean isAligned) {
// Check if event explicitly requires alignment
if (AlignmentUtil.requiresAlignment(event)) {
LOGGER.debug("Event {} requires aligned processing", event.getClass().getSimpleName());
return true;
}

// Fall back to the original isAligned parameter
return isAligned;
}

private void startTask() {
long start = System.currentTimeMillis();
this.executorService = Executors.getExecutorService(1, WORKER_FORMAT + context.getTaskId() + "-%d",
ComponentUncaughtExceptionHandler.INSTANCE);
executorService.execute(new WorkerTask());
LOGGER.info("taskId {} start task cost {}ms", context != null ? context.getTaskId() : "null", System.currentTimeMillis() - start);
}

public void alignedProcess(long fetchCount) {
LOGGER.debug("Executing aligned process for taskId {} with fetchCount {}", context.getTaskId(), fetchCount);
super.process(fetchCount, true);
}

public void unalignedProcess() {
try {
InputMessage input = inputReader.poll(DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
if (input != null) {
long windowId = input.getWindowId();
if (input.getMessage() != null) {
PipelineMessage message = input.getMessage();
processMessage(windowId, message);
} else {
long totalCount = input.getWindowCount();
processBarrier(windowId, totalCount);
}
}
} catch (Throwable t) {
if (running) {
LOGGER.error(t.getMessage(), t);
throw new GeaflowRuntimeException(t);
} else {
LOGGER.warn("service closed {}", t.getMessage());
}
}

if (!running) {
LOGGER.info("{} worker terminated",
context == null ? "null" : context.getTaskId());
}
}

/**
* Process message event and trigger worker to process.
*/
@Override
protected void processMessage(long windowId, PipelineMessage message) {
processMessageEvent(windowId, message);
}

/**
* Trigger worker to call processor finish.
*/
@Override
protected void processBarrier(long windowId, long totalCount) {
long processedCount = windowCount.containsKey(windowId) ? windowCount.get(windowId) : 0;
finishBarrier(totalCount, processedCount);
LOGGER.info("taskId {} windowId {} process total {} messages", context.getTaskId(), windowId, processedCount);
windowCount.remove(windowId);
}

/**
* Verify the processed count and total count, and whether the window id
* currently processed is consistent with the window id in the context.
*/
protected void finishBarrier(long totalCount, long processedCount) {
if (totalCount != processedCount) {
LOGGER.error("taskId {} {} mismatch, TotalCount:{} != ProcessCount:{}",
context.getTaskId(), totalCount, totalCount, processedCount);
}
context.getEventMetrics().addShuffleReadRecords(totalCount);

long currentWindowId;
try {
currentWindowId = processingWindowIdQueue.poll(DEFAULT_TIMEOUT_MS, TimeUnit.MILLISECONDS);
} catch (InterruptedException e) {
throw new GeaflowRuntimeException(e);
}
finish(currentWindowId);

// Current window id must be in [context.getCurrentWindowId() - 1, context.getCurrentWindowId()].
if (currentWindowId != context.getCurrentWindowId() && currentWindowId != context.getCurrentWindowId() - 1) {
String errorMessage = String.format("currentWindowId is %d from queue, id is %d from context",
currentWindowId, context.getCurrentWindowId());
LOGGER.error(errorMessage);
throw new GeaflowRuntimeException(errorMessage);
}
super.init(currentWindowId + 1);
}

@Override
public void interrupt() {
super.interrupt();
if (executorService != null) {
ExecutorUtil.shutdown(executorService);
executorService = null;
}
if (alignmentManager != null) {
alignmentManager.shutdown();
}
}

@Override
public void close() {
super.close();
this.running = false;
this.processingWindowIdQueue.clear();
this.windowCount.clear();
if (executorService != null) {
LOGGER.info("shutdown unaligned worker");
ExecutorUtil.shutdown(executorService);
executorService = null;
}
if (alignmentManager != null) {
alignmentManager.shutdown();
}
}

public class WorkerTask implements Runnable {

@Override
public void run() {
try {
while (running) {
unalignedProcess();
}
} catch (Exception e) {
LOGGER.error("Unaligned process encounter exception ", e);
throw new GeaflowRuntimeException(e);
}
}
}
}
```

**Specific tasks**:

1. **Performance optimization**

- Optimize the synchronization mechanism of alignment processing to reduce unnecessary waiting time

- Implement batch optimization of alignment processing

2. **Fault tolerance mechanism**

- Add a retry mechanism for alignment processing failures

- Implement a downgrade processing strategy for alignment timeouts

3. **Comprehensive testing**

- Unit testing: Verify the alignment processing logic of LoadGraphProcessEvent

- Integration testing: Test the behavioral consistency under different worker types

- Performance testing: Evaluate the impact of alignment processing on system performance

## Implementation plan

### Week 1: Current Status Research and Design Optimization
- In-depth analysis of existing code to identify all relevant components
- Design a complete alignment processing architecture
- Develop detailed interface specifications

### Week 2: Core Functionality Development
- Implement complete alignment processing logic for LoadGraphProcessEvent
- Improve relevant methods in AbstractUnAlignedWorker
- Update relevant factory and configuration classes

### Week 3: Integration and Testing
- Conduct comprehensive unit and integration tests
- Performance benchmarking and optimization
- Documentation updates and code review

### Week 4: Deployment and Verification
- Deploy and verify in a test environment
- Conduct stress and stability testing
- Prepare for production release

Start by reading the TODO in AbstractUnAlignedWorker and the alignment check in AbstractIterationComputeCommand.execute(). Then trace LoadGraphProcessEvent construction in SchedulerEventBuilder and compare the aligned and unaligned worker paths. The issue is complete only when the intended alignment behavior and supporting framework are defined, implemented, and covered by tests.

Written by the indexing model from the issue text.