apache / apache/dolphinscheduler

[DSIP-96][Worker] Unified Task Unique Identifier Design

Open
#17,752 3 comments 2 reactions 1 assignee Claimed by @det101 View on GitHub
discussion DSIP
Dominant language
Java
Stars
14.5k
Forks
5.1k
Avg merge
1d 21h
Merged PRs (30d)
29

Description

### Search before asking

- [x] I had searched in the [DSIP](https://github.com/apache/dolphinscheduler/issues/14102) and found no similar DSIP.

### Motivation

**Background:**

In DolphinScheduler, when tasks are submitted to external systems (such as Yarn, K8s), a unique identifier is required. Currently, there are several issues:

1. Insufficient uniqueness: User-defined `appName` may be duplicated and cannot guarantee uniqueness.
2. Fault tolerance difficulty: When Worker fails, it is impossible to accurately judge whether the task is still running outside based only on the identifier.
3. Non-unified implementation: Each task type (Spark, MapReduce, Flink, K8s) uses a different logic for unique id generation.

**Goals:**
- Guarantee uniqueness: Unique identifier generation should be based on `taskInstanceId` (globally unique).
- Preserve user intent: User-set `appName`/task name should be part of the identifier for usability/readability.
- Unified specification: All task types should follow the same identifier generation rule.
- Fault tolerance friendly: The format should help future features such as querying & handling running tasks for failover.

**Principles:**
- Uniqueness comes first even if readability is slightly affected.
- Backward compatibility: Should not break existing execution flow or user parameters.
- Extensible: Adapt different naming requirements from target systems.

### Design Detail

### Unique Identifier Generation Rules

#### 1. Yarn Tasks (Spark, MapReduce, Flink)
Rule:
- If user sets `appName`: `{userAppName}-{taskInstanceId}`
- Otherwise: `ds-{taskInstanceId}`
Examples:
- appName="my-spark-app", taskInstanceId=100 → `my-spark-app-100`
- (no appName), taskInstanceId=200 → `ds-200`

#### 2. K8s Tasks
Rule:
- Format: `{taskName}-{taskInstanceId}`
- `taskName` must be normalized (lowercase, remove special characters)
Examples:
- taskName="MyTask", taskInstanceId=100 → `mytask-100`

#### 3. Other Task Types
Rule:
- Format: `ds-task-{taskInstanceId}`
Example:
- taskInstanceId=100 → `ds-task-100`

### Features
- **Global Uniqueness:** Guaranteed by using taskInstanceId.
- **Traceability:** Easy to correlate with database and UI.
- **Readability:** Contains user’s input for easier search/identification.
- **Friendly for future fault tolerance:** Easy to query task and status by unified format.

### Implementation Plan

#### Phase 1: Utility Classes
- Add `TaskUniqueIdentifierGenerator` utility (Java)
- Add `TaskUniqueIdentifierExtractor` utility for parsing/extraction
- Write unit tests

#### Phase 2: Spark Integration
- Modify `SparkTask.populateSparkOptions()` to use the new generator
- Verify via tests

#### Phase 3: MapReduce Integration
- Modify `MapReduceArgsUtils.buildArgs()` to use the new generator
- Verify via tests

#### Phase 4: Flink Integration
- Modify `FlinkArgsUtils.buildRunCommandLine()` and `FlinkArgsUtils.buildInitOptionsForSql()`
- Verify via tests

#### Phase 5: K8s Task Optimization
- Modify `K8sTaskExecutor.buildK8sJob()`
- Ensure normalized format
- Verify via tests

#### Phase 6: Docs & Integration Tests
- End-to-end & performance testing
- Documentation improvement

### Recommended code interfaces:
```java
public interface TaskUniqueIdentifierGenerator {
String generateYarnId(String userAppName, long taskInstanceId);
String generateK8sName(String taskName, long taskInstanceId, int maxLength);
String generateDefaultId(long taskInstanceId);
}
public interface TaskUniqueIdentifierExtractor {
public static Optional extractTaskInstanceIdFromYarnAppName(String appName);
public static Optional extractTaskInstanceIdFromK8sName(String k8sName);
}
```

### Compatibility, Deprecation, and Migration Plan

Backward compatibility:
- Existing parameters and execution flows are not affected.
- User customized appName/task name is still included in identifier format.

Migration:
- Can be rolled out by task type or feature flag for stepwise adoption.
- No need for migration of historical tasks/data.

Deprecation:
- No existing feature deprecated.
- All new identifier formats are backward-compatible and fall back to old logic if new fails.

### Test Plan

- Unit tests for utility classes covering edge cases and normalization.
- Integration tests for each task type (Spark, Flink, MapReduce, K8s) verifying task names generated and used in external system submission.
- End-to-end validation: submit sample workflows and check that generated identifiers appear in Yarn/K8s UI/API.
- Performance tests to ensure no extra latency.
- Docs update with usage, formats, and troubleshooting.

### Code of Conduct

- [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct)

Contributor guide

Open the contributing guide

Assessment

This issue has not been assessed yet.

Get new issues in your inbox

A short digest of beginner-friendly GitHub issues.