dijiekstra commented on issue #18627:
URL:
https://github.com/apache/dolphinscheduler/issues/18627#issuecomment-5615293074
## 📋 Quick Reference of Core Modification Points
### Master-side changes (< 500 lines)
#### Modification 1: DependencyResolver integration
**Location**:
`dolphinscheduler-master/src/main/java/.../service/DependencyResolverService.java`
(new)
```java
public interface DependencyResolver {
DependencyStatus resolveDependency(Long workflowCode, Long taskCode,
String dependencyGroup);
void registerAssetStateChangeListener(AssetStateChangeListener listener);
}
```
#### Modification 2: `ASSET_SENSOR` task type
**Location**:
`dolphinscheduler-task-plugin/dolphinscheduler-task-asset-sensor/` (new module)
- `AssetSensorParameters extends AbstractParameters`
- `AssetSensorLogicTask extends AbstractLogicTask`
- `AssetSensorTracker implements TaskTracker`
#### Modification 3: dependency check in `TaskProcessor`
**Location**:
`dolphinscheduler-master/src/main/java/.../processor/TaskProcessor.java` (about
5 lines changed)
```java
// Add before sending to Worker in submitTask()
if (task.getType() == TaskType.ASSET_SENSOR) {
if (!dependencyResolver.resolveDependency(...).isReady()) {
return; // Keep WAITING_DEPENDENCY state
}
}
```
#### Modification 4: workflow startup check in `DAGExecutionEngine`
**Location**: `dolphinscheduler-master/.../engine/DAGExecutionEngine.java`
(about 10 lines changed)
After the workflow instance is created and before the DAG is generated,
workflow-level asset dependencies can optionally be validated in advance.
#### Modification 5: Master configuration and compensation scanner
**Location**:
`dolphinscheduler-master/.../scanner/AssetEventCompensationScanner.java` (new)
Cycle: every 5–15 minutes
Responsibilities:
- Backfill missed Paimon snapshot events
- Retry stuck `trigger_history` records
---
## 🗂️ Database table checklist
| Table | Key Fields | Unique Constraint | Indexes |
|-----|--------|--------|------|
| t_ds_asset | assetKey, catalogName, databaseName, tableName | uk_asset_key
| idx_asset_key |
| t_ds_asset_event | eventId, sourceType, assetKey, snapshotId |
uk_dedup(source, asset, event_type, snapshot) | idx_asset_snapshot,
idx_event_type_time |
| t_ds_asset_state | assetKey, latestSnapshotId, qualityStatus |
uk_asset_key | idx_asset_key |
| t_ds_asset_dependency | workflowCode, taskCode, assetKey, dependencyGroup
| - | idx_asset_key, idx_workflow |
| t_ds_asset_trigger_history | triggerKey, workflowCode, taskCode |
uk_trigger_key | idx_status_time |
| t_ds_asset_event_consumer_offset | sourceType, assetKey, consumerGroup |
uk_offset | - |
---
## 🔍 Key design points
### 1. `trigger_key` composition rule
```
workflow_{workflowCode}_group_{dependencyGroup}_snapshot_{id1}_{id2}_...
```
- Identifies one complete trigger decision
- Unique constraint guarantees idempotency
- No distributed lock required
### 2. Three-layer dependency evaluation
### 3. Idempotency guarantee
- The same `trigger_key` can be inserted successfully at most once
- Database unique constraints naturally handle concurrency races
- Compensation retries are automatically deduplicated
---
## ⚠️ Detailed explanation of 8 identified risks
| # | Risk | Impact | Mitigation |
|----|-----|------|--------|
| 1 | Backfill semantics | Should historical snapshots trigger? | Design a
`backfill_mode` marker |
| 2 | Multi-Master race | `trigger_history` insert conflict |
`uk_trigger_key` constraint + exception capture |
| 3 | Task polling pressure | DB query pressure | Cache + event-driven
callbacks to replace polling |
| 4 | CommandType compatibility | Existing logic may break | Search all
usages across the repo and adapt |
| 5 | Cross-project permissions | Permission model unclear | Design
separately later |
| 6 | Push/Poll deduplication | Clock skew / ordering differences | Version
comparison + unique constraints |
| 7 | Exactly-once scope | Over-promising semantics | Clarify the boundary:
`trigger_history` layer only |
| 8 | Mixed partition dependencies | Expression unclear | Add clearer rules
later |
---
## 📦 Test coverage checklist
### Unit tests
- [ ] Idempotency: the same `trigger_key` cannot be created twice
- [ ] Deduplication: duplicate events are rejected by database constraints
- [ ] Out-of-order handling: old snapshots cannot overwrite new snapshots
- [ ] AND/OR: combined evaluation of multi-asset dependencies
### Integration tests
- [ ] Snapshot advancement flow: event → state → dependency ready → trigger
→ DAG execute
- [ ] Multi-Master scenario: idempotency under concurrency
- [ ] Compensation flow: backfilling missed events, retrying stuck triggers
### End-to-end tests
- [ ] GMV scenario: complete ODS → DWD → DWS → ADS chain
- [ ] Performance test: response time under concurrent snapshot advancement
- [ ] Failure test: recovery from Master outage, network jitter, and similar
scenarios
---
## 🚀 Suggested focus points for community discussion
1. **Scope of the paradigm shift**
- Should all Lakehouse scenarios adopt the pure snapshot model?
- Should mixed time + snapshot triggering be allowed?
2. **Master modification risk assessment**
- Are the five modification points sufficient? Are there hidden
dependencies not yet covered?
- Will the one-line check in `TaskProcessor#submitTask()` affect
performance?
3. **Performance and observability**
- Is the 5–15 minute cycle for `CompensationScanner` reasonable?
- Do the 22+ metrics cover the key decision path?
4. **Ecosystem and extension**
- MVP supports only Paimon; how should the adaptation cost for
Iceberg/Hudi be estimated?
- How should authentication and rate limiting for Push event reception be
designed?
---
💬 **Comments and discussion are welcome!**
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
