[Fix][Connector-V2] Support schema evolution in Redis sink - #11646
[Fix][Connector-V2] Support schema evolution in Redis sink#11646lm-ylj wants to merge 8 commits into
Conversation
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for working on this! I fetched the head locally (b07583d98b), read the full diff, and traced the schema-change path end to end through both the Zeta and Flink sink runtimes. The approach makes sense and the structure of the change is genuinely good — the flush-before-refresh ordering and the checkpointed schema are exactly the right primitives. I did find a few things that I think should be addressed before merging; details below, and I'm happy to re-review as soon as you push an update.
What Problem Does This PR Solve?
User pain point. When a MySQL-CDC (or any CDC) source is wired to the Redis sink with format = json / format = text whole-row serialization, the sink builds its SerializationSchema once, from the SeaTunnelRowType captured at construction time, and never rebuilds it. After an upstream ALTER TABLE ... ADD COLUMN, the source starts emitting wider rows, but the sink keeps serializing with the original row type: newly added columns silently disappear from the Redis value, and after a DROP COLUMN the field names and the positional values drift out of alignment. The user sees stale or misaligned JSON in Redis with no error anywhere in the logs.
Fix approach. RedisSinkWriter now implements SupportSchemaEvolutionSinkWriter. On each schema change event it flushes everything buffered under the old schema, folds the event into the current TableSchema via TableSchemaChangeEventDispatcher, recomputes the physical row type, and rebuilds the serializer. The writer's state type changes from Void to TableSchema so the evolved schema survives checkpoint/restore, with RedisSink.restoreWriter rebuilding the writer from the restored schema.
One-sentence summary. The Redis sink stops freezing its serializer at job start and instead follows the live CDC schema, with the evolved schema persisted in checkpoint state.
1. Code Change Review
1.1 Core Logic Analysis
Precise description of the changes. Two production files carry the whole behavioral change:
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java— state typeVoid→TableSchema, plusrestoreWriterandgetWriterStateSerializer.seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java—SupportSchemaEvolutionSinkWriter, mutable schema/serializer fields,applySchemaChange,snapshotState.
Before:
public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, Void> ... {
private final SeaTunnelRowType seaTunnelRowType;
...
this.seaTunnelRowType = catalogTable.getSeaTunnelRowType();public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow, Void>
implements SupportMultiTableSinkWriter<Void> {
private final SeaTunnelRowType seaTunnelRowType;
private final SerializationSchema serializationSchema;After (RedisSinkWriter.java:61-69, 112-117):
public class RedisSinkWriter extends AbstractSinkWriter<SeaTunnelRow, TableSchema>
implements SupportMultiTableSinkWriter<Void>, SupportSchemaEvolutionSinkWriter {
private TableSchema tableSchema;
private SeaTunnelRowType seaTunnelRowType;
private SerializationSchema serializationSchema;
@Override
public void applySchemaChange(SchemaChangeEvent event) {
flush();
tableSchema = schemaChangeEventDispatcher.reset(tableSchema).apply(event);
seaTunnelRowType = tableSchema.toPhysicalRowDataType();
serializationSchema = createSerializationSchema(redisParameters, seaTunnelRowType);
}Key findings.
- The normal path does reach the changed logic, but only on Zeta.
SinkFlowLifeCycle.processSchemaChangeEvent(seatunnel-engine/seatunnel-engine-server/src/main/java/org/apache/seatunnel/engine/server/task/flow/SinkFlowLifeCycle.java:450-457) dispatches purely onwriter instanceof SupportSchemaEvolutionSinkWriter, so it never consults theSink-side interface. That is why the new E2E passes withoutRedisSinkdeclaring anything. On Flink the gate is different — see Issue 1. - The constructor switch from
catalogTable.getSeaTunnelRowType()tocatalogTable.getTableSchema()+toPhysicalRowDataType()is behavior-preserving:CatalogTable.getSeaTunnelRowType()(seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/CatalogTable.java:158-160) is literallytableSchema.toPhysicalRowDataType(). Existing non-CDC Redis jobs are unaffected. - The flush ordering is correct.
flush()(RedisSinkWriter.java:333-338) drainskeyBuffer/valueBuffer, which already hold serialized strings produced under the old schema, so draining before swapping the serializer is exactly right — no buffered row is ever re-serialized with a schema it did not belong to. - Checkpoint state is genuinely serializable.
MultiTableSink.getWriterStateSerializer()(seatunnel-api/.../multitablesink/MultiTableSink.java:232-234) usesDefaultSerializer(Java serialization), andTableSchemainheritsSerializablefromAbstractSchema(seatunnel-api/src/main/java/org/apache/seatunnel/api/table/catalog/AbstractSchema.java:35) withColumn implements Serializable(Column.java:38). So the new state round-trips correctly. - Not overriding
getPhysicalSinkTableIdentifier()is the right call here — the defaultOptional.empty()keeps the legacy source-only routing, and sinceapplySchemaChangehas no external side effect on Redis (no DDL is executed), sibling sub-writers sharing a Redis instance do not need the fan-out that JDBC-style sinks need. Good judgement.
In-depth correctness analysis. The change takes effect for whole-row serialization only: getValue(...) falls back to serializationSchema.serialize(element) only when handleHashType/handleOtherTypes return null. For data_type = hash with an explicit hash_key_field/hash_value_field, and for configured value_field, the serializer is never consulted, so schema evolution is a no-op on those paths. Your documentation says this correctly, which I appreciate. It does not take effect for field names embedded in key / custom key placeholders — also correctly documented.
Complete runtime path (Zeta, multi-table CDC):
MySQL-CDC source emits AlterTableAddColumnEvent
-> SinkFlowLifeCycle.received(record) [SinkFlowLifeCycle.java:234-239]
-> processSchemaChangeEvent(event) [SinkFlowLifeCycle.java:450-457]
-> writer instanceof SupportSchemaEvolutionSinkWriter -> applySchemaChange(event)
-> MultiTableSinkWriter.applySchemaChange(event)
-> hasSourceMatchedWriter(event) // routes by source table id
-> enqueueSchemaChangeBarrier(event) // parks every queue worker
-> dispatchSchemaChangeToTargets(...)
-> applySchemaChangeToTarget(...) [MultiTableSinkWriter.java:507-527]
-> RedisSinkWriter.applySchemaChange(event) [RedisSinkWriter.java:112-117]
-> flush() [RedisSinkWriter.java:333-338]
-> doBatchWrite() // old-schema rows land in Redis first
-> TableSchemaChangeEventDispatcher.reset(tableSchema).apply(event)
-> seaTunnelRowType = tableSchema.toPhysicalRowDataType()
-> serializationSchema = createSerializationSchema(...) [RedisSinkWriter.java:278-300]
Next data row
-> RedisSinkWriter.write(element) [RedisSinkWriter.java:97-109]
-> fields = seaTunnelRowType.getFieldNames() // now the evolved names
-> getValue(element, fields) -> serializationSchema.serialize(element)
Checkpoint
-> RedisSinkWriter.snapshotState(cpId) [RedisSinkWriter.java:329-331]
-> Collections.singletonList(tableSchema.copy())
-> MultiTableSinkWriter.snapshotState -> MultiTableState{SinkIdentifier -> List<TableSchema>}
-> MultiTableSink.getWriterStateSerializer -> DefaultSerializer [MultiTableSink.java:232-234]
Restore
-> MultiTableSink.restoreWriter(context, states) [MultiTableSink.java:169-192]
-> sink.restoreWriter(proxy, state)
-> RedisSink.restoreWriter(context, states) [RedisSink.java:61-73]
-> new RedisSinkWriter(restoredSchema, redisParameters)
Flink streaming path (this is where the gap is)
-> SinkExecuteProcessor.createVersionSpecificDataStreamSink(...)
[seatunnel-core/seatunnel-flink-starter/seatunnel-flink-starter-common/.../SinkExecuteProcessor.java:57]
-> if (isStreaming && sink instanceof SupportSchemaEvolutionSink) { insert BroadcastSchemaHandler }
-> RedisSink is NOT SupportSchemaEvolutionSink -> operator is never inserted
1.2 Compatibility Impact
Fully compatible. No option is renamed or removed, no default changes, and the initial row type is derived identically to before (point 2 above). The writer state type moves from Void to TableSchema, but the previous implementation returned no writer-state serializer at all, so a job restored from a pre-upgrade checkpoint arrives at RedisSink.restoreWriter with an empty/absent state list and correctly falls through to createWriter(context) (RedisSink.java:62-65). Serialization format, protocol, and public API are untouched. No incompatible-changes.md entry is needed.
1.3 Performance / Side-Effect Analysis
Negligible and correctly scoped. applySchemaChange runs once per DDL, not per row; the extra work is one TableSchema mutation, one toPhysicalRowDataType(), and one serializer construction. snapshotState adds one TableSchema.copy() per checkpoint per sub-writer — a small object graph, no hot-path cost. write() still allocates Arrays.asList(seaTunnelRowType.getFieldNames()) per row, but that predates this PR and is unchanged. No new locks, no new network calls, no resource that needs releasing. The extra flush at DDL time is required for correctness and happens at a natural boundary.
1.4 Error Handling and Logging
This is the weakest area of the change, and it is where most of my findings sit.
Issue 1: RedisSink does not implement SupportSchemaEvolutionSink
-
Location:
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java:36-37 -
Problem description: The writer side of the contract is implemented, but the sink side is not. Every other schema-evolution sink in the repo declares both halves —
JdbcSink.java:330-336,PaimonSink.java:214-220,DorisSink.java:174-180,StarRocksSink.java:128-134,ElasticsearchSink.java:110-112,ConsoleSink.java:64-70— each returning its supportedSchemaChangeTypelist. Two concrete consequences follow. First, on Flink streaming,SinkExecuteProcessor.createVersionSpecificDataStreamSinkinserts theBroadcastSchemaHandleroperator only whensink instanceof SupportSchemaEvolutionSink(flink-starter-commonSinkExecuteProcessor.java:57, flink-20:56, flink-13:213). Without it, the DDL reaches whichever parallel sink subtask happens to receive it while the sibling subtasks keep their stale serializer. Second,MultiTableSink.supports()(MultiTableSink.java:386-392) delegates to the first sub-sink and returnsCollections.emptyList()when that sub-sink does not implement the interface — so the job's machine-readable answer to "does this sink support schema evolution?" is "no", while the docs added in this PR list Redis as supported. -
Potential risk: On Flink with
parallelism > 1, rows written after the DDL carry a mix of old-schema and new-schema JSON depending on which subtask handled them — arguably worse than the uniformly-stale behavior before this PR, because it is inconsistent rather than merely outdated. Nothing in the code prevents a user from running this configuration; only a sentence in the docs does. -
Best improvement: Add the interface and declare the types the dispatcher can actually handle.
TableSchemaChangeEventDispatcher.createHandlers()registers handlers for add/modify/drop/change column events, so the honest list is:public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, TableSchema> implements SupportMultiTableSink, SupportSchemaEvolutionSink { @Override public List<SchemaChangeType> supports() { return Arrays.asList( SchemaChangeType.ADD_COLUMN, SchemaChangeType.DROP_COLUMN, SchemaChangeType.RENAME_COLUMN, SchemaChangeType.UPDATE_COLUMN); } }
Note this is also enforced in one direction by
ConnectorSpecificationCheckTest.checkSupportSchemaEvolutionSink(seatunnel-dist/src/test/java/org/apache/seatunnel/api/connector/ConnectorSpecificationCheckTest.java:210-232) — sincecreateWriteralready returnsRedisSinkWriter, that check will pass as soon as you add the interface. If you would rather keep the scope strictly Zeta-only for now, the alternative is to keep the sink undeclared and remove Redis fromdocs/{en,zh}/introduction/configuration/schema-evolution.md, but I think adding the interface is clearly the better outcome. -
Severity: High
-
Raised by another reviewer: No
Issue 2: restoreWriter fails the job on any state divergence, with no diagnostic context
- Location:
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSink.java:61-73 - Problem description: When several old subtask states are merged into one writer on a parallelism downscale, the code compares every
TableSchemaand throwsnew IOException("Redis sink restored inconsistent table schema states")on the first mismatch. The message carries no table identifier, no checkpoint id, no column counts, and no indication of which two states disagreed, so an operator hitting this on a production restore has nothing to act on. Because this runs on the restore path, the job cannot start at all — it is not a recoverable per-record failure. - Potential risk: A restore-time hard failure with an undiagnosable message. Given that DDL is applied to all sub-writers through the same barrier, divergence should be rare, but "rare and fatal with no context" is the worst combination for on-call debugging.
- Best improvement: Option A (preferred): include the differing schemas' field names and the sink table identifier in the message, e.g.
"Redis sink restored inconsistent table schema states for <tableId>: [id,name] vs [id,name,email]". Option B: instead of failing, log aWARNand pick the schema with the largest column set, on the reasoning that the widest schema is the most recent one and the serializer will simply emitnullfor fields the incoming rows do not carry. Option A is safer and I would go with it; Option B trades a hard failure for a soft one and would need its own test. - Severity: Medium
- Raised by another reviewer: No
Issue 3: new public methods and mutable fields have no documentation
- Location:
RedisSinkWriter.java:66-69(fieldstableSchema,seaTunnelRowType,serializationSchema),RedisSinkWriter.java:112-117(applySchemaChange),RedisSinkWriter.java:119-129(toTableSchema),RedisSinkWriter.java:329-331(snapshotState),RedisSink.java:61-77(restoreWriter,getWriterStateSerializer) - Problem description: Three fields changed from
finalto mutable and none of them says why, when they are reassigned, or which thread may observe them.applySchemaChangeis the heart of the feature and does four ordered operations — flush, fold, recompute, rebuild — where the ordering is load-bearing, but nothing records that the flush must precede the serializer swap.restoreWriterimplements a non-obvious "all states must agree" policy with no comment explaining the invariant. The project's guidance asks for comments on methods and non-trivial fields that carry business logic, state transitions, boundary conditions, or protocol semantics; all of these qualify. - Potential risk: The next person to touch
applySchemaChangecan easily reorder the flush below the serializer rebuild — a change that compiles, passes the existing unit tests that assert on final state, and silently corrupts every row still buffered at DDL time. - Best improvement: Add short Javadoc to
applySchemaChangestating that buffered rows are serialized under the previous schema and must be drained first; documentsnapshotState/restoreWriteras the checkpointed-schema contract; and add a one-line field comment noting that these three fields are reassigned on schema change. - Severity: Medium
- Raised by another reviewer: No
Issue 4: E2E covers only one of the four serialization configurations, and no real restore
- Location:
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/mysqlcdc_to_redis_with_schema_change.conf:35-44 - Problem description: The job config pins
data_type = key,format = json,parallelism = 1. The serializer-refresh path is also reachable withformat = text(TextSerializationSchema,RedisSinkWriter.java:287-292) and withdata_type = hashwhenhash_key_field/hash_value_fieldare unset andhandleHashTypereturnsnull, falling through to whole-row serialization. Neither is exercised anywhere. Separately,testSnapshotStateRestoresLatestSchema(RedisSinkWriterTest.java:244-253) verifies the restore contract by constructing a new writer directly from the state object; it never goes throughRedisSink.restoreWriter, so thestates.size() > 1comparison branch and thenull/empty-states branch added in this PR have zero coverage. - Potential risk: The
data_type = hashcase has a real behavioral wrinkle worth pinning down in a test: RedisHSETis additive, so after aDROP COLUMNthe removed field remains on already-existing hash keys even though it is no longer written. Your docs phrase this carefully as "dropped fields are no longer written", which is accurate, but a test would lock the behavior in. The uncoveredrestoreWriterbranches are the ones most likely to fire in production and least likely to be exercised by hand. - Best improvement: Add unit tests for
RedisSink.restoreWritercoveringnullstates, empty states, one state, and two divergent states. For the E2E, a second@TestTemplatecase withformat = textwould be cheap since the containers are already up; adata_type = hashcase is nice-to-have rather than required. - Severity: Medium
- Raised by another reviewer: No
Issue 5: toTableSchema fabricates column metadata and is now unreachable from production code
- Location:
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java:119-129 - Problem description: It builds every column as
PhysicalColumn.of(name, type, 0L, true, null, null)— column length0, always nullable, no default, no comment — and drops the primary key and constraint keys entirely. After this PR,RedisSinkalways uses theTableSchemaconstructor (RedisSink.java:57,:72), so the only remaining callers of theSeaTunnelRowTypeconstructor are the unit tests. - Potential risk: Low today, but this fabricated schema is now a checkpointed object. If any future code path routes through the legacy constructor, a schema with zeroed lengths and a dropped primary key gets written into checkpoint state and restored as though it were authoritative.
- Best improvement: Either mark the
SeaTunnelRowTypeconstructor@Deprecatedwith a note that it exists for test compatibility, or update the tests to build aTableSchemadirectly and delete both the constructor andtoTableSchema. The tests already have aninitialSchema()helper (RedisSinkWriterTest.java:303), so the second option is a small edit. - Severity: Low
- Raised by another reviewer: No
Issue 6: mutable cross-thread fields are not volatile
- Location:
seatunnel-connectors-v2/connector-redis/src/main/java/org/apache/seatunnel/connectors/seatunnel/redis/sink/RedisSinkWriter.java:66-69 - Problem description: Under
SupportMultiTableSinkWriter,write()runs onMultiTableWriterRunnableworker threads whileapplySchemaChangeis invoked from the barrier callback, potentially on a different thread. I tracedenqueueSchemaChangeBarrier(MultiTableSinkWriter.java) and it does park every worker on a shared barrier before dispatching, which should establish the necessary happens-before edge — so I am not claiming a demonstrable race here. But the guarantee lives entirely in engine code that the connector does not control, andwrite()reads these fields outside any monitor (flush()issynchronized,write()is not). - Potential risk: If the engine's barrier implementation is ever relaxed, this becomes a silent visibility bug that produces mixed-schema output with no error — the hardest class of bug to diagnose in a data pipeline.
- Best improvement: Mark the three fields
volatile. The cost is a handful of memory barriers per row, which is noise next to the Redis round trip, and it makes the writer's correctness self-contained. - Severity: Low
- Raised by another reviewer: No
2. Code Quality Assessment
2.1 Coding Standards
Naming, package layout, and the use of TableSchemaChangeEventDispatcher all follow existing connector conventions closely — this reads like code that belongs in the project. The one standards gap is documentation coverage on the newly added methods and the newly mutable fields, raised as Issue 3 above. No wildcard imports, no System.out.println, no license-header problems.
2.2 Test Coverage and Test Stability
Coverage. The unit tests are well chosen: add-column refresh, drop-column refresh, flush-before-refresh ordering, and state round-trip. The E2E covers the realistic MySQL-CDC to Redis path with two rows and both an add and a drop. The gaps are the restoreWriter branches and the text/hash configurations, raised as Issue 4.
Stability rating: Stable.
Evidence for the rating:
RedisSchemaChangeIT.awaitRedisJson(RedisSchemaChangeIT.java:271-284) usesAwaitility.await().atMost(180s).pollInterval(2s).untilAsserted(...)rather than a fixedThread.sleep, and re-checksassertJobHasNotFinished(jobFuture)on every poll so a crashed job fails fast with the exit code instead of timing out after three minutes. That is exactly the right shape.- Container readiness is explicit:
HostPortWaitStrategywith a 2-minute startup timeout (RedisSchemaChangeIT.java:115-117),Startables.deepStart(...).join(), and an eagerassertEquals("PONG", jedis.ping())(:128) so a half-started Redis fails instartUprather than mid-assertion. - No hard-coded host ports —
redisContainer.getFirstMappedPort()(:126) — so parallel CI lanes cannot collide. - State is reset on both ends:
jedis.del(KEY_1, KEY_2)instartUp(:150) and again during cleanup (:245), and cleanup failures are merged rather than swallowed. - The unit tests are Mockito-based and fully deterministic: no sleeps, no clock reads, no randomness, no ordering dependencies, fresh writer per
@Test.
One minor observation, not a blocker: waitForIncrementalRead (RedisSchemaChangeIT.java:286-301) gates on server log content (serverLogs.contains(INCREMENTAL_READ_MARKER)), which is a log-keyword readiness signal rather than a data-level one. It is mitigated by also asserting the captured table name and by the fact that every subsequent assertion is data-level, so I would leave it as is — just be aware it is coupled to that log line's wording.
2.3 Documentation Updates
docs/en and docs/zh are both updated and consistent with each other, and Redis is added to both schema-evolution support lists. I want to call out that the documentation is unusually precise here — explicitly stating that Redis executes no DDL, that only whole-row JSON/TEXT serialization is refreshed, that key / value_field / hash_key_field / hash_value_field are not rewritten, and that upscaling parallelism after a DDL is unsupported. That is exactly the level of honesty that saves users from filing bugs against documented limitations. The one thing to reconcile is the Zeta-only scoping versus the sink contract in Issue 1.
3. Architectural Soundness
3.1 Elegance of the Solution
Precise fix. It changes the one thing that was actually wrong — a serializer frozen at construction — and reuses the framework's existing TableSchemaChangeEventDispatcher rather than hand-rolling event handling. It does not reach into the engine or add Redis-specific machinery to shared code.
3.2 Maintainability
Good, with the caveat in Issue 3. The applySchemaChange method is four readable lines, but three of the four orderings are load-bearing and none is documented. Adding those comments would make this comfortably maintainable.
3.3 Extensibility
The pattern here — hold a TableSchema, fold events through the dispatcher, rebuild derived objects, checkpoint the schema — is the same shape other schemaless sinks would need. Nothing is hard-coded to Redis beyond createSerializationSchema, so this is a reasonable template for the next connector.
3.4 Historical-Version Compatibility
Compatible. Restoring a job checkpointed by an older Redis sink lands in the empty-states branch of restoreWriter and falls through to createWriter, and the initial row type derivation is byte-for-byte equivalent to the previous code path. No upgrade action is required from users.
4. Issue Summary
| No. | Issue | Location | Severity |
|---|---|---|---|
| 1 | RedisSink does not implement SupportSchemaEvolutionSink; Flink broadcast operator never inserted and MultiTableSink.supports() reports empty |
RedisSink.java:36-37 |
High |
| 2 | restoreWriter throws on state divergence with no table/schema context |
RedisSink.java:61-73 |
Medium |
| 3 | New methods and newly mutable fields lack documentation | RedisSinkWriter.java:66-69,112-129,329-331; RedisSink.java:61-77 |
Medium |
| 4 | No coverage for format = text, data_type = hash, or the restoreWriter branches |
mysqlcdc_to_redis_with_schema_change.conf:35-44; RedisSinkWriterTest.java:244-253 |
Medium |
| 5 | toTableSchema fabricates column metadata and is now production-unreachable |
RedisSinkWriter.java:119-129 |
Low |
| 6 | Mutable cross-thread fields are not volatile |
RedisSinkWriter.java:66-69 |
Low |
5. Merge Recommendation
Conclusion: Ready to merge after fixes
1. Blockers — must be fixed
- Issue 1 (High) — Add
implements SupportSchemaEvolutionSinkwith asupports()list toRedisSink. This is the declaration every peer sink makes, it is what gates the FlinkBroadcastSchemaHandleroperator, and without it the docs added in this PR claim support that the code does not declare. It is a small change and it closes the only path in this PR that can produce inconsistent output. - Issue 2 (Medium) — Add table/schema context to the
restoreWriterfailure message. This one is cheap and it is the difference between a diagnosable and an undiagnosable production restore failure. - Issue 3 (Medium) — Add documentation to the new methods and the newly mutable fields, in particular the flush-before-swap ordering in
applySchemaChange.
2. Recommended fixes — non-blocking
- Issue 4 (Medium) — I would like to see unit coverage for the
restoreWriterbranches, since they are new logic with none today. The extra E2E configurations are genuinely optional; I am not asking you to grow CI time for them. - Issue 5 (Low) and Issue 6 (Low) — Both are hygiene. Take them if you are touching those lines anyway.
Overall assessment. This is a well-scoped, well-documented change that fixes a real and fairly subtle data-correctness problem, and the test work behind it is above average for a connector PR — the E2E in particular is one of the more carefully written ones I have read recently, with no timing anti-patterns at all. The only structural thing missing is the sink-side half of the schema-evolution contract; once that is in, the feature is declared as consistently as the other sinks and works on the Flink path too.
Is there a better alternative implementation? Not meaningfully. Option A — what you have done, holding the TableSchema in the writer and rebuilding the serializer — is the approach JdbcSink and PaimonSink take and is the right one. Option B would be to make the serializer resolve field indices lazily per row so no rebuild is ever needed, but that pushes per-row cost into the hot path to avoid a once-per-DDL cost, which is the wrong trade. Stay with Option A.
On CI: the required Build check on b07583d98b is still running at the time of this review, so I have no CI verdict to report yet. The branch is diverged from dev (ahead_by=3, behind_by=3); I am recording that as a mergeability fact only and I am not asking you to sync, since there is no failing signal to attribute to upstream drift. If Build comes back red on something unrelated to your diff, ping me and I will help narrow it down.
Thanks again for the care you put into the docs and the limitation notes — please push an update when you get a chance and I will re-review the full head.
|
Addressed the requested review changes in
|
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Thanks for the quick turnaround, and for the detailed changelog in your comment — it made this re-review straightforward to verify. I re-fetched the current head, re-read the full diff against dev from scratch (not just what changed since my last pass), and re-traced the runtime path end to end again rather than trusting my previous notes. All three blockers from my last review are genuinely fixed, and I did not find anything I should have caught the first time that is still lingering — no carryover apology owed here, just a clean follow-up.
What Problem Does This PR Solve?
User pain point. When a CDC source (MySQL-CDC or similar) feeds the Redis sink with whole-row format = json / format = text serialization, the sink used to build its SerializationSchema once from the SeaTunnelRowType captured at construction time and never rebuild it. After an upstream ALTER TABLE ... ADD COLUMN / DROP COLUMN, the sink kept serializing with the stale row type — new columns silently disappeared, and after a drop the field/value alignment drifted — with no error surfaced anywhere.
Fix approach. RedisSinkWriter implements SupportSchemaEvolutionSinkWriter: on each schema-change event it flushes everything buffered under the old schema, folds the event into the current TableSchema via TableSchemaChangeEventDispatcher, recomputes the physical row type, and rebuilds the serializer. The schema is checkpointed (writer state type Void → TableSchema) so it survives restart. As of this commit, RedisSink also implements SupportSchemaEvolutionSink, which is what gates the Flink BroadcastSchemaHandler operator and what MultiTableSink.supports() reports — this was the missing half in the previous commit.
One-sentence summary. The Redis sink now follows the live CDC schema on both engines instead of freezing it at job start, with the evolved schema persisted in checkpoint state.
1. Code Change Review
1.1 Core Logic Analysis
What changed since my last review (b07583d98b → c01219d4f9fb). Four files: RedisSink.java (+34/-…), RedisSinkWriter.java (+22 doc/deprecation), RedisSinkTest.java (new, 212 lines), RedisSinkWriterTest.java (assertions rewritten, constructor calls migrated). I re-read every one of these plus every file that was already correct before (RedisClient, RedisParameters, the E2E conf/IT, the docs) to make sure nothing regressed from the interaction between the old and new code.
Issue 1 (previously High) — fixed. RedisSink now declares the sink-side half of the contract:
public class RedisSink extends AbstractSimpleSink<SeaTunnelRow, TableSchema>
implements SupportMultiTableSink, SupportSchemaEvolutionSink {
...
@Override
public List<SchemaChangeType> supports() {
return Arrays.asList(
SchemaChangeType.ADD_COLUMN,
SchemaChangeType.DROP_COLUMN,
SchemaChangeType.RENAME_COLUMN,
SchemaChangeType.UPDATE_COLUMN);
}
}This is the exact same list JdbcSink, PaimonSink, DorisSink, StarRocksSink, ElasticsearchSink, and ConsoleSink declare, and it matches what TableSchemaChangeEventDispatcher.createHandlers() actually routes (add/modify/drop/change-column handlers). With this in place, SinkExecuteProcessor.createVersionSpecificDataStreamSink will insert the BroadcastSchemaHandler on the Flink streaming path, and MultiTableSink.supports() no longer silently reports emptyList() while the docs claim support. RedisSinkTest.testAdvertisesSupportedSchemaChanges pins this down directly. Good fix, and it closes the one path in the previous commit that could have produced inconsistent per-subtask output on Flink.
Issue 2 (previously Medium) — fixed. restoreWriter now reports table identity and both diverging field lists instead of a bare message:
throw new IOException(
String.format(
"Redis sink cannot restore writer for table %s because state 0 fields %s differ from state %d fields %s",
catalogTable.getTablePath().getFullName(),
schemaFields(restoredSchema),
stateIndex,
schemaFields(state)));This is exactly the diagnostic content I asked for — table path, which state index disagreed, and the actual field lists on both sides. RedisSinkTest.testRestoreWriterRejectsConflictingStatesWithTableAndFieldContext asserts on the message content, not just that an exception is thrown, which is the right level of test rigor for a message whose entire value is being readable.
Issue 3 (previously Medium) — fixed. The three schema-derived fields, applySchemaChange, snapshotState, and restoreWriter all now carry Javadoc that states the invariant, not just what the code does:
/**
* Flushes all rows serialized with the previous schema before refreshing the writer's
* schema-derived views. The engine orders this callback with row processing, so all three views
* are updated before any subsequent row is processed.
*/
@Override
public void applySchemaChange(SchemaChangeEvent event) {This is the comment that would have stopped the exact reordering mistake I was worried about — it says outright that the flush must happen before the fields are reassigned, and why.
Issue 5 (previously Low) — fixed. The SeaTunnelRowType constructor and toTableSchema are now @Deprecated with a pointer to the TableSchema constructor. The schema-evolution-specific tests (testAddColumnUsesLatestSchemaForJsonSerialization, testDropColumnRemovesFieldFromJsonSerialization, testSchemaChangeFlushesOldRowsBeforeSerializerRefresh, testSnapshotStateRestoresLatestSchema) were migrated to build a TableSchema directly via initialSchema(); the plain custom-key tests still use the deprecated constructor, which is a reasonable place to leave it since those tests don't exercise schema evolution at all.
Issue 6 (previously Low, "not volatile") — not implemented as literal volatile, but I accept the author's reasoning. From the update comment: "Independent volatile writes would still not atomically publish the correlated Schema/RowType/Serializer views, so they would not make the transition self-contained." That is correct — three independent volatile fields give per-field visibility but not atomicity across the triple, so volatile alone would not actually close the gap I was gesturing at. The ordering contract is now documented on the fields and the callback instead. I re-traced the actual concurrency path below and I'm now willing to say more confidently than last time that this is safe in practice, not just "probably fine."
Deeper concurrency trace (this is the part of the "review scope" I want to go further on than last time, since re-review is exactly the moment to stress-test an assumption instead of re-stating it). RedisSinkWriter.applySchemaChange is reached only through MultiTableSinkWriter.applySchemaChange → enqueueSchemaChangeBarrier, which enqueues a SchemaChangeRequest onto every queue and blocks on SchemaChangeBarrier.awaitCompletion(). Every MultiTableWriterRunnable.run() processes queue elements inside synchronized (this), and SchemaChangeRequest.process() calls reachBarrier(), which does not return for any participating worker — including the worker that owns the queue holding the target Redis writer — until the last-arriving worker has finished running dispatchSchemaChangeToTargets (the actual RedisSinkWriter.applySchemaChange call) and called completed.countDown(). MultiTableSinkWriter.snapshotState() reads a writer's state under synchronized (runnable.get(i)) for that writer's own queue index i; since worker i cannot release that same monitor until the barrier (including the schema mutation, wherever in the fan-out it physically runs) has fully completed, the CountDownLatch plus the monitor re-acquisition together establish a real happens-before edge from the schema mutation to the checkpoint read, regardless of which worker thread actually executed the mutation. I looked for a gap where checkQueueRemain()'s pre-check (isProcessingRow() is false for schema-change requests, so a barrier in flight isn't counted as "pending work") could let snapshotState() race ahead — but that pre-check is only a busy-wait optimization; the actual safety net is the per-queue synchronized block taken afterward, which does not have this gap. I could not construct a scenario where a checkpoint or a subsequent row write observes a schema mutation that is only partially applied. This confirms my original assessment was right, and gives me more confidence in it than "I am not claiming a demonstrable race here" did last time.
Compatibility of the constructor change (re-verified, unchanged from before). catalogTable.getTableSchema() + toPhysicalRowDataType() is still byte-for-byte equivalent to the old catalogTable.getSeaTunnelRowType() path (CatalogTable.getSeaTunnelRowType() is literally tableSchema.toPhysicalRowDataType()), so non-CDC Redis jobs are unaffected. Checkpoint state remains genuinely serializable (TableSchema → AbstractSchema implements Serializable, Column implements Serializable), and TableSchema/PhysicalColumn use Lombok @EqualsAndHashCode(callSuper = true), so the restoreWriter divergence check compares actual field content, not object identity — I checked this specifically because the whole safety of Issue 2's fix depends on .equals() being structural rather than reference-based.
Runtime path (unchanged shape from before, now correct on both engines):
CDC source emits AlterTableAddColumnEvent
-> SinkFlowLifeCycle.processSchemaChangeEvent(event)
-> writer instanceof SupportSchemaEvolutionSinkWriter -> applySchemaChange(event)
-> MultiTableSinkWriter.applySchemaChange(event)
-> enqueueSchemaChangeBarrier(event) // parks every queue worker behind a shared barrier
-> dispatchSchemaChangeToTargets(...)
-> RedisSinkWriter.applySchemaChange(event)
-> flush() // old-schema rows land in Redis first
-> tableSchema = dispatcher.reset(tableSchema).apply(event)
-> seaTunnelRowType = tableSchema.toPhysicalRowDataType()
-> serializationSchema = createSerializationSchema(...) // fresh instance, fresh field order
Next data row -> write(element) -> getValue(...) -> serializationSchema.serialize(element)
Checkpoint -> snapshotState(cpId) -> Collections.singletonList(tableSchema.copy())
Restore -> RedisSink.restoreWriter(context, states) -> new RedisSinkWriter(restoredSchema, redisParameters)
Flink streaming (previously the gap, now closed)
-> SinkExecuteProcessor.createVersionSpecificDataStreamSink(...)
-> sink instanceof SupportSchemaEvolutionSink == true -> BroadcastSchemaHandler inserted
1.2 Compatibility Impact
Fully compatible. No option renamed or removed, no default changed. The writer-state type change (Void → TableSchema) is handled: a job restored from a pre-upgrade checkpoint has no writer state at all, so restoreWriter correctly falls through its states == null || states.isEmpty() branch to createWriter(context). Adding SupportSchemaEvolutionSink only turns on machinery that was previously inert for Redis (the Flink broadcast handler and MultiTableSink.supports() reporting); it does not change behavior for jobs that never send schema-change events. No incompatible-changes.md entry is needed.
1.3 Performance / Side-Effect Analysis
Unchanged from my last assessment and still correct: applySchemaChange runs once per DDL, not per row; snapshotState adds one small TableSchema.copy() per checkpoint per sub-writer; write()'s per-row Arrays.asList(...) allocation predates this PR. No new locks, no new network calls introduced by this commit's changes specifically (RedisSink.java, RedisSinkWriter.java diffs are Javadoc/interface/exception-message only on top of the already-reviewed logic).
1.4 Error Handling and Logging
Issue 1: E2E and hash-type coverage gaps remain (carried forward from previous Issue 4, downgraded).
- Location:
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/mysqlcdc_to_redis_with_schema_change.conf:35-44(unchanged since my last review); no new hash-drop test inRedisSinkWriterTest.java. - Problem description: The restore-path gap I raised last time (
RedisSink.restoreWriter'snull/empty/single/multi-state branches) is now fully covered by the six new tests inRedisSinkTest.java— that was the part "most likely to fire in production and least likely to be exercised by hand," and it's done. What's still open is purely the E2E/unit breadth: the E2E config still pinsdata_type = key, format = json, parallelism = 1, soformat = textanddata_type = hash(wherehandleHashTypefalls through to whole-row serialization whenhash_key_field/hash_value_fieldare unset) are still only exercised by inspection of the code, not by a running test. The hashHSET-is-additive behavior after aDROP COLUMN(the removed field remains on already-written hash keys) is documented accurately but still not pinned down by a test. - Potential risk: Low. This was already the non-blocking half of the previous review, and the higher-risk half (restore branches) is now covered.
- Best improvement: A second
@TestTemplateE2E case withformat = textwould be cheap since the containers are already up; adata_type = hashcase is nice-to-have. Not required for this PR. - Severity: Low
- Raised by another reviewer: No
No other Issues found in this round. The three blockers and one of the two non-blocking items from my last review are resolved; nothing new turned up in re-reading the full diff, and my deeper concurrency trace above (going further than "should establish the necessary happens-before edge") did not find a race.
2. Code Quality Assessment
2.1 Coding Standards
Same clean baseline as before — no wildcard imports, no System.out.println, license headers intact, naming and package layout consistent with the rest of the connector. The documentation gap I raised (Issue 3) is closed: every new mutable field, applySchemaChange, snapshotState, and restoreWriter now carry Javadoc that explains the invariant, not just a restatement of the code.
One purely cosmetic observation, not a defect: RedisSinkWriterTest.java's schema-evolution tests were rewritten from field-by-field ObjectNode assertions (json.get("email").asText(), json.has("legacy_note"), etc.) to whole-string equality against the serialized JSON ("{\"id\":1,\"name\":\"Alice\",\"email\":...}"). I checked whether this trades away robustness for brittleness: JsonSerializationSchema builds a new instance (with a fresh, uninitialized ObjectNode) every time createSerializationSchema is called in applySchemaChange, and ObjectNode.put() preserves insertion order for pre-existing keys while appending genuinely new keys — so given a fixed TableSchema column order, the emitted key order is deterministic, not an artifact of hash iteration. The new assertions are consistent with that and are not flaky; they're just more sensitive to future changes to key ordering than the old structural checks were. Not something I'd ask you to change.
2.2 Test Coverage and Test Stability
Coverage. RedisSinkTest.java (new) directly closes the restore-branch gap I flagged: null states, empty states, single state, identical merged states, conflicting states (with message-content assertions), and a full serialize → deserialize → restore → write round trip that proves the checkpoint bytes actually produce a working writer, not just an equal-looking schema object. Combined with the existing RedisSinkWriterTest.java (add/drop/flush-ordering/snapshot), this is solid coverage of the new logic's primary paths.
Stability rating: Stable. I re-checked the E2E (RedisSchemaChangeIT.java, unchanged since my last review) against the flaky-test checklist again from scratch rather than trusting my prior notes: Awaitility.await().atMost(180s).pollInterval(2s) with assertJobHasNotFinished re-checked on every poll (no Thread.sleep anywhere in the assertion path), explicit container readiness via HostPortWaitStrategy + eager PONG check, no hard-coded ports (getFirstMappedPort()), state reset on both ends of each test. The new unit tests in RedisSinkTest.java are Mockito-based, deterministic, no sleeps or ordering dependencies. Same rating as before, re-verified rather than carried over unchecked.
2.3 Documentation Updates
docs/en and docs/zh for both Redis.md and schema-evolution.md are unchanged since my last review and were already accurate and unusually careful about limitations (no DDL executed, which fields are/aren't rewritten, parallelism-upscale-after-DDL not supported). One thing worth noting, not a blocker: the docs still say "Redis Sink supports schema evolution with SeaTunnel Zeta" and don't mention Flink, even though this commit's fix to Issue 1 is exactly what turns on the Flink BroadcastSchemaHandler gate. I think leaving the doc scoped to Zeta is the right call for now, since this PR doesn't add a Flink-engine E2E test to actually prove the Flink path end-to-end — but if a Flink test is added in a follow-up, the doc should be updated to say so explicitly rather than leaving it silently broader than documented.
3. Architectural Soundness
3.1 Elegance of the Solution (Precise fix)
Unchanged assessment: this reuses the framework's TableSchemaChangeEventDispatcher rather than hand-rolling event handling, and now declares both halves of the SupportSchemaEvolutionSink(Writer) contract consistently with every other schema-evolution sink in the repo.
3.2 Maintainability
Improved concretely since last review: the load-bearing orderings (flush-before-swap, restore-must-agree) are now documented next to the code that enforces them, which was the main maintainability gap I raised.
3.3 Extensibility
Unchanged: the pattern (hold a TableSchema, fold events through the dispatcher, rebuild derived objects, checkpoint the schema) remains a reasonable template for the next schemaless-sink connector that needs this.
3.4 Historical-Version Compatibility
Unchanged and re-verified: a job checkpointed by a pre-upgrade Redis sink restores through the empty-states branch into createWriter, and the initial row-type derivation is identical to the previous code path.
4. Issue Summary
| No. | Issue | Location | Severity |
|---|---|---|---|
| 1 | E2E covers only data_type=key/format=json; format=text and data_type=hash whole-row paths, and the hash DROP COLUMN retention behavior, are undocumented-by-test (restore-path gap from the previous review is now closed) |
mysqlcdc_to_redis_with_schema_change.conf:35-44 |
Low |
Previously-open Issues now resolved and not re-listed: Issue 1 High (SupportSchemaEvolutionSink missing) — fixed; Issue 2 Medium (restoreWriter diagnostics) — fixed; Issue 3 Medium (missing documentation) — fixed; Issue 5 Low (toTableSchema production-unreachable) — fixed via @Deprecated. Previous Issue 6 Low (fields not volatile) — not implemented literally, but resolved via a correct technical argument plus documentation; I verified the underlying happens-before chain independently and did not find a gap.
5. Merge Recommendation
Conclusion: Ready to merge
1. Blockers — must be fixed
None. All three blocking issues from my previous review are genuinely fixed, not just documented around.
2. Recommended fixes — non-blocking
- Issue 1 (Low) — Optional follow-up E2E coverage for
format = textanddata_type = hash. Nice to have, not required.
Overall assessment. This is now a complete, consistent implementation of both halves of the schema-evolution contract, with the diagnostics and documentation that make it maintainable by someone other than the original author. I want to be explicit that I went back and independently re-derived the concurrency safety argument for the checkpoint/schema-change interaction rather than re-stating my previous conclusion, specifically because that is exactly the kind of thing a second commit can quietly break — it didn't, and I now have a more concrete argument for why than I did last time.
CI status. The required Build check is currently failing on c01219d4f9fb. I traced it to the fork's own Actions run (unit-test (11, windows-latest) job) rather than trusting the Apache-side pointer, and the actual failure is Exception calling "DownloadFile" ... The remote server returned an error: (429) Too Many Requests while fetching the Maven wrapper jar, followed by Could not find or load main class org.apache.maven.wrapper.MavenWrapperMain. That is a transient upstream rate-limit on the Windows runner's Maven-wrapper bootstrap, not a compile or test failure related to this diff — nothing in the log references Redis, schema evolution, or any file touched by this PR. I'd recommend re-running that specific failed job rather than pushing a new commit; I don't consider this a code-level blocker, but it should go green before this is actually merged.
Is there a better alternative implementation? No change from my last review: holding the TableSchema in the writer and rebuilding derived objects on schema change is the same approach JdbcSink/PaimonSink use, and remains the right one over a lazily-resolving-per-row alternative that would push cost into the hot path.
Thanks again for turning this around quickly and for the precise summary of what changed — it made verifying each fix straightforward rather than something I had to reverse-engineer from the diff alone.
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Re-review at the current head. Short version: nothing changed in this PR's own files since my last pass — the only new commit is a merge from dev that pulled in the HugeGraph connector, a MaxCompute tweak, and a Knowledge Sync metadata field, none of which touch anything under connector-redis or the schema-evolution docs. I diffed my last-reviewed commit against 3b18b5281d restricted to this PR's own file list and it's empty. So the diff itself is a clean no-op and my previous "all three blockers fixed" conclusion still holds for the code.
However, checking CI on the current head (as I always do before signing off) turned up something genuinely new that I had not seen before, and it changes my recommendation.
What Problem Does This PR Solve?
User pain point. When a CDC source (MySQL-CDC or similar) feeds the Redis sink with whole-row format = json / format = text serialization, the sink used to build its SerializationSchema once from the SeaTunnelRowType captured at construction time and never rebuild it. After an upstream ALTER TABLE ... ADD COLUMN / DROP COLUMN, the sink kept serializing with the stale row type — new columns silently disappeared, and after a drop the field/value alignment drifted — with no error surfaced anywhere.
Fix approach. RedisSinkWriter implements SupportSchemaEvolutionSinkWriter: on each schema-change event it flushes everything buffered under the old schema, folds the event into the current TableSchema via TableSchemaChangeEventDispatcher, recomputes the physical row type, and rebuilds the serializer. RedisSink implements SupportSchemaEvolutionSink so the Flink BroadcastSchemaHandler gate and MultiTableSink.supports() both see it. The schema is checkpointed (writer state type Void → TableSchema) so it survives restart.
One-sentence summary. Unchanged from before — the Redis sink now follows the live CDC schema on both engines instead of freezing it at job start.
1. Code Change Review
1.1 Core Logic Analysis
What's actually new since my last review. Nothing in RedisSink.java, RedisSinkWriter.java, the docs, or the test/E2E files this PR owns. git diff <my-last-reviewed-commit>..3b18b5281d -- <this PR's files> is empty; the only content that landed is an unrelated Merge branch 'dev' that touches 128 files across HugeGraph, MaxCompute, and a metadata-transform module. I confirmed none of those paths overlap with Redis or docs/*/introduction/configuration/schema-evolution.md.
What I did check again anyway. Since the merge changed the module's neighbors even if not the module itself, I re-ran the correctness claim I made last time — that catalogTable.getTableSchema().toPhysicalRowDataType() (new) and catalogTable.getSeaTunnelRowType() (old) are byte-for-byte equivalent — directly against CatalogTable.java:158-160 on the current head:
public SeaTunnelRowType getSeaTunnelRowType() {
return tableSchema.toPhysicalRowDataType();
}Still literally the same call. That part of my earlier assessment stands unchanged.
The new finding (this is the part that's genuinely unreviewed). I checked gh pr checks on 3b18b5281d the way I always do before approving, and the required Build check is failing. I traced it into the fork's own run rather than trusting the Apache-side pointer (lm-ylj/seatunnel run 31066415889, job unit-test (8, ubuntu-latest)). This is not the Windows Maven-wrapper rate-limit issue I called out and dismissed last time — that was a different job on a different commit. This is a real failure inside connector-redis itself:
[ERROR] Tests run: 9, Failures: 0, Errors: 9, Skipped: 0 - in org.apache.seatunnel.connectors.seatunnel.redis.Redis5Test
[ERROR] Tests run: 9, Failures: 0, Errors: 9, Skipped: 0 - in org.apache.seatunnel.connectors.seatunnel.redis.Redis7Test
All 18 test methods across both classes (Redis5Test/Redis7Test, both subclasses of RedisTemplateTest) fail with a bare java.lang.NullPointerException, each one at the exact line of its own SinkFlowTestUtils.runBatchWithCheckpointDisabled(...) call (RedisTemplateTest.java:134,148,169,181,193,207,233,247,262), with sub-10ms elapsed time per test. @BeforeAll container startup clearly succeeds (each class reports its own 3.7s/2.7s total, consistent with a real container boot, not a class-level init failure), so this is failing deep inside the RedisSinkFactory → RedisSink → RedisSinkWriter construction/write path that this PR's two production files sit on — not in test infrastructure.
Two things rule out "this is just flaky CI":
- It's 100% reproducible, not intermittent — 9/9 in Redis 5, 9/9 in Redis 7, same failure mode, same near-instant timing in both.
- These same tests pass cleanly on
dev. I pulled the logs from the last known-greenBuildrun onapache/seatunnel:dev(2026-07-08, run28910769257, jobunit-test (8, ubuntu-latest)):Redis5TestandRedis7Testboth reportTests run: 9, Failures: 0, Errors: 0there, with essentially identical class-level timings (3.684s / 2.465s vs. this run's 3.787s / 2.663s). Same suite, same container images, same runner type — the only variable is this PR's diff.
I also want to be precise about what this isn't: it isn't something either of my previous reviews should have caught. At c01219d4f9fb (what I actually approved), this exact job — unit-test (8, ubuntu-latest) — was cancelled, not failure, because the matrix's Windows leg died first and fail-fast killed the rest before Ubuntu got to run Redis5Test/Redis7Test. Across this PR's entire history, this is the first time that job has ever completed on this branch, and it's red. Nobody has seen this signal before.
I was not able to pin down the exact null site from the console log — Surefire's summary reporter here prints only the exception class with no at ... frames, and no surefire-reports artifact was uploaded for this run — so I can't hand you a file:line the way I could for the issues in my last review. What I can say with confidence: it is inside the code path this PR added or changed (RedisSink's SPI-driven construction via RedisSinkFactory, or RedisSinkWriter's constructor/write/flush), since that is the only thing that differs between the green dev run and this red run for these two test classes, and the row-type-derivation equivalence I re-verified above rules out the one theory I had going in.
1.2 Compatibility Impact
Unchanged from my last review and still correct on its own terms: no option renamed/removed, no default changed, writer-state type change (Void → TableSchema) falls through correctly for pre-upgrade checkpoints, SupportSchemaEvolutionSink only turns on previously-inert machinery. No incompatible-changes.md entry needed. This assessment is orthogonal to the CI finding above — a regression that breaks Redis5Test/Redis7Test at construction time is a correctness bug, not a compatibility break, but it does mean I can no longer certify that non-CDC, everyday Redis sink usage (exactly what RedisTemplateTest's scenarios represent: plain KEY/LIST/SET/HASH/ZSET writes with no schema evolution involved at all) still works on this head.
1.3 Performance / Side-Effect Analysis
No new information since my last review for the reviewed diff itself. Not applicable to the new finding since I don't have a confirmed root cause to reason about side effects for.
1.4 Error Handling and Logging
Issue 1: Required Build check is red on connector-redis's own pre-existing test suite, and this is the first time this test job has run to completion on this PR
- Location:
seatunnel-connectors-v2/connector-redis/src/test/java/org/apache/seatunnel/connectors/seatunnel/redis/RedisTemplateTest.java(lines noted above), exercisingRedisSink.java/RedisSinkWriter.javain their current-head form. - Problem description: See the trace above. 18/18 tests across
Redis5Test/Redis7Testfail with an immediateNullPointerExceptionfrom inside the sink-construction/write path, on a test suite that is unmodified by this PR and passes cleanly ondev. - Potential risk: If this reflects a real defect (which the evidence strongly suggests, given 100% reproducibility and a clean baseline comparison), it means the basic, non-CDC Redis sink path — not just the new schema-evolution path — is broken on this head. That would be a regression far more serious than anything flagged in my previous two reviews.
- Best improvement: Re-run
mvn -pl seatunnel-connectors-v2/connector-redis test -Dtest=Redis5Test,Redis7Testwith full stack traces (-DtrimStackTrace=falseor inspect thetarget/surefire-reports/*.txtfiles) to get the actual throw site, then fix it. Given the symptom profile, I'd start by checking whetherRedisSinkFactory.createSink(context).createSink()followed bycreateWriter(...)populates every fieldRedisSinkWriter's constructor now depends on when driven through the realTableSinkFactoryContextpath (as opposed to the hand-builtTableSchema/mocks used inRedisSinkTest/RedisSinkWriterTest), and whether anything in the newRedisSinkTest.javaleaves shared/static state behind that the immediately-followingRedis5Test/Redis7Testfork depends on. - Severity: High (blocking)
- Raised by another reviewer: No
No other issues found. Everything I flagged in my first review — the missing SupportSchemaEvolutionSink declaration, the undiagnostic restoreWriter failure message, the missing Javadoc — remains fixed and unregressed; I re-read all four files again in full to be sure the dev-drift merge didn't quietly touch anything (it didn't).
2. Code Quality Assessment
2.1 Coding Standards
Unchanged from my last review: no wildcard imports, no System.out.println, license headers intact, naming/package layout consistent, documentation gap from my first review remains closed.
2.2 Test Coverage and Test Stability
This is where my rating changes. My previous "Stable" rating was based on RedisSchemaChangeIT (E2E) and the Mockito-based RedisSinkTest/RedisSinkWriterTest — all of which I still consider stable and deterministic; nothing about them has changed and nothing in this round's investigation contradicts that. But "stable" for a connector also has to include the module's existing, previously-green test suite continuing to pass, and Redis5Test/Redis7Test are failing 100% of the time on this head while passing 100% of the time on dev. That is a regression, not flakiness — flaky tests fail intermittently under the same code; this fails deterministically and only appeared once the job was finally allowed to run to completion. I'm downgrading the stability verdict for this PR as a whole to Not stable — real, reproducible test regression, root cause not yet isolated.
2.3 Documentation Updates
Unchanged and unaffected by this finding — docs/en and docs/zh for Redis.md and schema-evolution.md are identical to what I reviewed and approved last time, and the content is still accurate for the schema-evolution behavior specifically. It says nothing about the basic non-CDC write path being broken, which is exactly the gap the new finding exposes — if the root cause turns out to affect ordinary Redis sink usage, the docs won't need a correction, the code will.
3. Architectural Soundness
3.1 Elegance of the Solution / 3.2 Maintainability / 3.3 Extensibility
Unchanged from my last review.
3.4 Historical-Version Compatibility
Unchanged and still correct in isolation (checkpoint/restore fallback logic is untouched by the new finding). But I can no longer say the sink "just works" for existing, non-evolving Redis jobs until Issue 1 is root-caused, since RedisTemplateTest's scenarios are exactly that: no CDC, no schema change, just a plain sink write.
4. Issue Summary
| No. | Issue | Location | Severity |
|---|---|---|---|
| 1 | Required Build check fails: Redis5Test/Redis7Test (18/18 tests, pre-existing suite, unmodified by this PR) fail with an immediate NullPointerException through the RedisSinkFactory → RedisSink → RedisSinkWriter path; same suite passes cleanly on dev baseline; this job never ran to completion on any earlier commit of this PR |
RedisTemplateTest.java:134,148,169,181,193,207,233,247,262, exercising RedisSink.java / RedisSinkWriter.java |
High (blocking) |
Previously-open issues (all from my first review, all confirmed still fixed and unregressed): missing SupportSchemaEvolutionSink declaration, undiagnostic restoreWriter message, missing documentation on schema-derived fields — not re-listed.
5. Merge Recommendation
Conclusion: Not ready to merge — new blocking CI failure, root cause not yet isolated
1. Blockers — must be fixed
- Issue 1 (High) — Root-cause and fix the
Redis5Test/Redis7Testfailures. This is not the same CI noise I waved off last time (that was a Windows-runner Maven-wrapper 429, unrelated to any file in this diff and on a job that has no bearing onconnector-redis's own tests). This is the Linux unit-test job that actually exercises this connector, finally running to completion for the first time on this branch, and it is red on a test suite this PR does not touch, in a way that reproduces 100% of the time and does not reproduce ondev. I'd treat this as a real regression until proven otherwise.
2. Recommended fixes — non-blocking
- Everything from my last review's "non-blocking" list (optional
format = text/data_type = hashE2E coverage) still stands and is still optional.
Overall assessment. I want to be straightforward about the sequence here: my last review approved this PR based on a complete, careful diff read and a documented CI check — and at that time, the CI check I could see was the Windows failure I correctly identified as unrelated noise. What I didn't have, and couldn't have had, was a completed Linux unit-test run, because that job had never once finished on this branch due to matrix fail-fast. Now that it has, it's telling us something real. I don't think this is a case of "approve now, chase CI later" — a 100%-reproducible NullPointerException in the connector's own pre-existing test suite, on the exact classes this PR modifies, immediately after those same classes' own new/modified unit tests passed, is worth understanding before merge rather than after. I'd ask for the actual stack trace (re-run with -DtrimStackTrace=false or pull target/surefire-reports/*.txt) so we can see the throw site directly rather than guessing between "a real construction-path bug" and "test-fixture state leaking from RedisSinkTest into the next fork" — both are plausible from what I can see in the diff, and the fix differs depending on which one it is.
CI status. Required Build check: failing on 3b18b5281d (unit-test (8, ubuntu-latest), fork run 31066415889, job 92505969597) — real failure inside connector-redis, not the previously-identified Windows infra issue. unit-test (11, ubuntu-latest) and both Windows legs were cancelled by fail-fast before running, so I have no signal on Java 11.
Is there a better alternative implementation? No change from my last review on this question — the schema-evolution approach itself remains the right one.
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Re-review at the current head (9b45c83e278d). Short version: the new commit since my last review is a real, targeted fix for the Redis5Test/Redis7Test regression I flagged last time, and I traced it down to the exact line rather than taking the fix on faith. I don't think there's a carryover apology owed here — my last review correctly identified the regression and correctly said the root cause wasn't yet isolated; this round is exactly where that gets isolated and closed.
What Problem Does This PR Solve?
User pain point. When a CDC source (MySQL-CDC or similar) feeds the Redis sink with whole-row format = json / format = text serialization, the sink used to build its SerializationSchema once from the SeaTunnelRowType captured at construction time and never rebuild it. After an upstream ALTER TABLE ... ADD COLUMN / DROP COLUMN, the sink kept serializing with the stale row type — new columns silently disappeared, and after a drop the field/value alignment drifted — with no error surfaced anywhere.
Fix approach. RedisSinkWriter implements SupportSchemaEvolutionSinkWriter: on each schema-change event it flushes everything buffered under the old schema, folds the event into the current TableSchema via TableSchemaChangeEventDispatcher, recomputes the physical row type, and rebuilds the serializer. RedisSink implements SupportSchemaEvolutionSink so the Flink BroadcastSchemaHandler gate and MultiTableSink.supports() both see it. The schema is checkpointed (writer state type Void → TableSchema) so it survives restart. This commit adds a normalization step so a TableSchema with a null constraint-key list can't reach the checkpoint-copy path and crash.
One-sentence summary. Unchanged from my last two reviews — the Redis sink now follows the live CDC schema on both engines instead of freezing it at job start.
1. Code Change Review
1.1 Core Logic Analysis
What's new since my last review. One commit, two files: RedisSinkWriter.java (+12/-2) and RedisSinkWriterTest.java (+14 new test). I confirmed this by diffing my last-reviewed commit (c01219d4f9fb) against the current head restricted to no filter at all — the rest of the 100+ file delta is dev-branch drift (HugeGraph, Maxcompute, file-connector, metadata-transform, docs) that doesn't touch connector-redis anywhere.
Root cause of last round's failure, isolated. Last time I could see 18/18 failures in Redis5Test/Redis7Test with a bare NullPointerException and no stack trace in the console output, and said I couldn't hand you a file:line. I went and got one this round by reading the full call chain instead of waiting for a fresh CI run to hand it to me:
RedisTemplateTest.getTableSchema()— the shared fixture bothRedis5TestandRedis7Testextend, unmodified by this PR — buildsnew TableSchema(getColumns(), null, null). The third argument,constraintKeys, isnullby construction (RedisTemplateTest.java:293-295).TableSchema's own constructor accepts this with no validation (TableSchema.java:37-42), so this is a legal, if unusual, object.- That schema flows through
CatalogTable.of(...)intoRedisSink(config, table)(RedisSink.java:46-51), andRedisSink.createWriter()passes it straight intonew RedisSinkWriter(tableSchema, redisParameters)(RedisSink.java:59-61). Before this commit, the constructor stored it as-is. SinkFlowTestUtils.runWriter()— the shared test harness everyRedisTemplateTestcase goes through viarunBatchWithCheckpointDisabled— computesneedsFinalCheckpoint = recordsSinceLastCheckpoint > 0 || checkpointState.triggeredCount == 0 || checkpointOptions.isTriggerOnFinish()(SinkFlowTestUtils.java:287-290). Because every one of these tests writes at least one row,recordsSinceLastCheckpoint > 0is always true, sotriggerCheckpoint(..., force=true)always runs at the end of the row loop — even in the "checkpoint disabled" tests — and that callssinkWriter.snapshotState(checkpointId)(SinkFlowTestUtils.java:326-327).RedisSinkWriter.snapshotState()isCollections.singletonList(tableSchema.copy())(RedisSinkWriter.java:361-363, unchanged by this commit).TableSchema.copy()doesconstraintKeys.stream().map(ConstraintKey::copy)...with no null guard (TableSchema.java:85-94, inseatunnel-api, not touched by this PR).
That's the exact throw site: a null-safe-looking test fixture from years before this PR, routed through a checkpoint call that fires even when checkpointing is nominally "disabled," into a shared API method that has never been null-safe. It also matches every symptom I saw last time: near-instant failure (this happens right after the last row is written, before any network I/O), 9/9 in both classes (every test method takes the exact same path), and "deep inside the RedisSinkFactory → RedisSink → RedisSinkWriter construction/write path" — it's actually the checkpoint path immediately following construction and write, which is close enough that my prediction holds up.
The fix, and why it's sufficient. RedisSinkWriter's constructor now does this.tableSchema = normalizeTableSchema(tableSchema) (RedisSinkWriter.java:100):
private static TableSchema normalizeTableSchema(TableSchema tableSchema) {
if (tableSchema.getConstraintKeys() != null) {
return tableSchema;
}
return TableSchema.builder()
.columns(tableSchema.getColumns())
.primaryKey(tableSchema.getPrimaryKey())
.build();
}TableSchema.Builder.constraintKeys is a new ArrayList<>() that's never assigned null (TableSchema.java:53), so .build() always produces a schema with a non-null (possibly empty) constraint-key list. I checked whether this guarantee can be undone later in the writer's life and it can't: the only other place this.tableSchema is reassigned is applySchemaChange (RedisSinkWriter.java:135), which goes through TableSchemaChangeEventDispatcher.apply() → AlterTableSchemaEventHandler. Every branch of that handler (applyAddColumn, applyDropColumn, applyModifyColumn) rebuilds the schema via TableSchema.builder()...constraintKey(schema.getConstraintKeys())...build(), and Builder.constraintKey(List) is this.constraintKeys.addAll(constraintKeys) on that same never-null list — so as long as the schema handed to the dispatcher already has a non-null list (guaranteed by the constructor normalization), every schema produced afterward keeps that guarantee by induction. There is exactly one place tableSchema can start life with a null constraint-key list — the constructor — and this commit closes it there. This is the right place to fix it: a single choke point, not a scattered set of null checks.
Compatibility of the fix itself. Converting null to an empty List<ConstraintKey> is semantically a no-op — nothing in this connector reads getConstraintKeys() for anything other than copy()'s null-unsafe iteration and TableSchema's generated equals()/hashCode() (used by restoreWriter's divergence check from my second review). Both a null list and an empty list mean "this table has no constraint keys" everywhere in this codebase; the normalization doesn't drop or fabricate any real constraint information, it just makes the "no constraints" case representable without also being unsafe to copy.
One thing worth flagging as a non-blocking observation, not a defect in this PR. CatalogTable.copy() (CatalogTable.java:135-142) also calls tableSchema.copy() unconditionally, so the same latent NPE in TableSchema.copy() exists as a systemic footgun in shared seatunnel-api code, not just in this connector — any table constructed with a null constraint-key list and later copied anywhere in the engine (not only via Redis's checkpoint path) would hit it. Fixing TableSchema.copy() itself to null-guard constraintKeys would close the whole class of bug at the source instead of one connector at a time, but that's seatunnel-api shared code and rightly out of scope for a Redis-connector bugfix PR. I'm noting it so it doesn't get lost, not asking for it here.
Runtime path (unaffected by this commit beyond the constructor normalization step, unchanged from my last two reviews otherwise):
RedisSinkFactory.createSink(context).createSink() -> new RedisSink(config, catalogTable)
-> RedisSink.createWriter(context) -> new RedisSinkWriter(tableSchema, redisParameters)
-> normalizeTableSchema(tableSchema) // new: null constraintKeys -> empty list, once
-> this.tableSchema = ... // guaranteed non-null constraintKeys from here on
CDC source emits AlterTableAddColumnEvent
-> SinkFlowLifeCycle.processSchemaChangeEvent(event) -> applySchemaChange(event)
-> flush() -> tableSchema = dispatcher.reset(tableSchema).apply(event) // still non-null, by induction
-> seaTunnelRowType = tableSchema.toPhysicalRowDataType()
-> serializationSchema = createSerializationSchema(...)
Checkpoint -> snapshotState(cpId) -> tableSchema.copy() // no longer NPEs: constraintKeys is never null here
Restore -> RedisSink.restoreWriter(context, states) -> new RedisSinkWriter(restoredSchema, redisParameters)
1.2 Compatibility Impact
Fully compatible, and this commit specifically restores compatibility with the plain (non-CDC) Redis sink path that the previous commit had accidentally broken. No option renamed or removed, no default changed. The normalization only affects the in-memory representation of constraintKeys (null → empty list); it has no effect on the serialized checkpoint format's shape beyond that same substitution, and DefaultSerializer (Java serialization) round-trips an empty ArrayList exactly as well as it round-trips null. Everything I verified as compatible in my last two reviews (writer-state type change, restoreWriter fallback for pre-upgrade checkpoints, row-type-derivation equivalence) is untouched by this commit and still holds.
1.3 Performance / Side-Effect Analysis
Negligible, and correctly scoped to construction time only. normalizeTableSchema runs once per writer construction (fresh create or restore), not per row or per checkpoint — it's a single getConstraintKeys() != null check that's true on every normal path (a real CatalogTable built through the SPI from a schema-aware source will almost always have a non-null, if possibly empty, constraint-key list already; null only shows up from hand-built test fixtures like RedisTemplateTest's). No new allocations on the hot path, no new locks, no new I/O.
1.4 Error Handling and Logging
No new issues in this commit's own diff. The one item still open from my last review is unchanged and remains non-blocking:
Issue 1 (carried over, Low, unchanged from last review): E2E and hash-type coverage gaps remain.
- Location:
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e/src/test/resources/mysqlcdc_to_redis_with_schema_change.conf:35-44(unmodified since I first raised this). - Problem description: The E2E config still pins
data_type = key, format = json, parallelism = 1;format = textanddata_type = hash(whole-row fallback) are still exercised only by code inspection, not by a running test. - Potential risk: Low, unchanged assessment from before.
- Best improvement: Optional follow-up E2E coverage; not required for this PR.
- Severity: Low
- Raised by another reviewer: No
No other issues found in the code itself.
CI, and why I'm not treating a green run as a formality this time. As of this review, the required Build check on 9b45c83e278d is still pending on the apache-side pointer. I traced into the fork's own run (lm-ylj/seatunnel run 31088899370) rather than trusting that pointer, the way I always do:
unit-test (11, windows-latest): failure. I pulled the full log and the actual failure isorg.apache.seatunnel.engine.server.dag.physical.StateTransitionCleanupTestthrowingjava.lang.IllegalStateException: Node failed to start!after a 311-second hang, insideseatunnel-engine-server— a module this PR does not touch, and a Hazelcast-node-startup failure that has nothing to do withconnector-redis, schema evolution, or checkpoint-state serialization. This is a different symptom than the Maven-wrapper 429 I dismissed as infra noise two reviews ago, but it's the same category: unrelated-module, environment-shaped failure.unit-test (8, ubuntu-latest),unit-test (11, ubuntu-latest),unit-test (8, windows-latest): all cancelled by the matrix's fail-fast once the Windows-11 leg failed. This meansunit-test (8, ubuntu-latest)— the exact job whoseRedis5Test/Redis7Testrun is what I need to see green to confirm this fix — has now failed to complete on two consecutive commits of this PR, for two unrelated reasons each time. I don't consider that a pattern in this PR's own code; it reads as this workflow's fail-fast matrix being unlucky twice in a row on the leg that happens to matter most for this specific regression.connector-redis-it (8/11, ubuntu-latest)(the E2E IT job, which is a different test surface than the unit-test job) was stillin_progressafter I waited roughly 12 minutes; I did not wait longer than that.
What this means for my conclusion. I traced the exact NPE call chain by hand (root cause above) rather than relying on a fresh CI run to hand it to me, and I'm confident in that trace independent of CI: the fix sits at the one place tableSchema can start life null-constraint, it's provably propagated correctly afterward, and it's covered by a unit test that exercises the exact previously-crashing call (snapshotState() on a writer built from a null-constraint-key schema). That said, I got burned two reviews ago by approving before the Ubuntu unit-test job had ever run to completion on this branch, so I'm being explicit rather than repeating that: I have not yet seen a completed, green unit-test (*, ubuntu-latest) run on this exact head. I recommend a job-level rerun of the cancelled unit-test (8, ubuntu-latest) / unit-test (11, ubuntu-latest) legs (or the whole matrix, if the fail-fast keeps eating them) once the current run finishes, specifically to get direct confirmation that Redis5Test/Redis7Test are green again — I expect them to be, based on the trace above, but "I expect" and "I saw green" are different claims and I want to be honest about which one this is.
2. Code Quality Assessment
2.1 Coding Standards
Consistent with the rest of the PR — no wildcard imports, no System.out.println, license headers intact. One gap worth naming, in the same spirit as the documentation issue I raised (and that got fixed) in my first review: normalizeTableSchema has no comment explaining why it exists. It's a private, three-line method, but the invariant it protects — "a TableSchema with a null constraint-key list must never reach TableSchema.copy()" — isn't obvious from the code alone and is exactly the kind of thing a future editor could silently undo (e.g., by adding a second place that assigns this.tableSchema without routing through this method). A one-line comment along the lines of "TableSchema.copy() does not null-check constraintKeys; normalize here so checkpointed schemas are always copy-safe" would make the invariant self-documenting the same way the flush-before-swap comment in applySchemaChange already does. Minor, not a blocker.
2.2 Test Coverage and Test Stability
Coverage. RedisSinkWriterTest.testSnapshotStateNormalizesMissingConstraintKeys (new) constructs a writer from new TableSchema(initialSchema().getColumns(), null, null) — the exact shape RedisTemplateTest.getTableSchema() uses — and asserts snapshotState() returns a schema with a non-null, empty constraint-key list. I checked that this test would have failed with a NullPointerException against the pre-fix code (it exercises tableSchema.copy() through the exact same null-unsafe path I traced above), so this isn't a test that merely restates the implementation — it's a real regression test for the specific bug. Combined with the existing RedisSinkTest/RedisSinkWriterTest suites from my last two reviews, this closes the coverage gap that mattered.
Stability rating: Stable, unchanged from my last review for the tests this PR owns — the new test is Mockito-based, deterministic, no sleeps, no ordering dependencies. This rating is about the tests this PR adds; it doesn't extend to the CI infrastructure flakiness discussed in section 1.4, which is a separate axis I've addressed there directly rather than folding into this rating.
2.3 Documentation Updates
Unchanged and unaffected — docs/en/docs/zh for Redis.md and schema-evolution.md are identical to what I reviewed and approved two rounds ago, and this commit is purely an internal correctness fix with no user-visible behavior change to document.
3. Architectural Soundness
3.1 Elegance of the Solution
The fix is precise: it normalizes at the single point where an unsafe value can enter the writer's state, rather than adding defensive null checks at every later use site (copy(), equals(), the dispatcher). That's the right shape for this kind of bug.
3.2 Maintainability
Good, modulo the missing one-line comment noted in 2.1 — the invariant being protected isn't self-evident from the code, and a maintainability win this specific here (an invariant that's easy to accidentally undo) is worth the one sentence.
3.3 Extensibility
Unchanged from my last review.
3.4 Historical-Version Compatibility
Unchanged and re-verified: this commit doesn't touch the pre-upgrade-checkpoint fallback path (restoreWriter's null/empty-states branch), and the normalization is purely additive to the writer's internal state handling.
4. Issue Summary
| No. | Issue | Location | Severity |
|---|---|---|---|
| 1 | E2E covers only data_type=key/format=json; format=text and data_type=hash whole-row paths remain untested by a running test (carried over, unchanged) |
mysqlcdc_to_redis_with_schema_change.conf:35-44 |
Low |
| 2 | normalizeTableSchema has no comment explaining the invariant it protects (TableSchema.copy() is not null-safe on constraintKeys) |
RedisSinkWriter.java:140-148 |
Low |
Previously-open Issues from earlier rounds, all confirmed still fixed and unregressed: missing SupportSchemaEvolutionSink declaration, undiagnostic restoreWriter message, missing documentation on schema-derived fields, toTableSchema/legacy-constructor deprecation. The High-severity CI regression from my last review (Redis5Test/Redis7Test NullPointerException) is fixed at the code level — root cause isolated to TableSchema.copy()'s null-unsafe constraintKeys iteration, closed by construction-time normalization — but not yet independently confirmed by a completed CI run, for the infrastructure reasons detailed in section 1.4.
5. Merge Recommendation
Conclusion: Ready to merge, pending one thing outside the code
1. Blockers — must be fixed
None in the code. Everything I've flagged across all three of my previous reviews is fixed and re-verified; this commit's own diff is a correct, minimal, well-tested fix for the one thing that was still open.
2. Recommended fixes — non-blocking
- Issue 1 (Low) — optional
format = text/data_type = hashE2E coverage, carried over from my first review. - Issue 2 (Low) — a one-line comment on
normalizeTableSchemaexplaining why it exists.
Overall assessment. I traced the NullPointerException I found last round to its exact origin — RedisTemplateTest's pre-existing, PR-unmodified fixture building a TableSchema with a null constraint-key list, routed through a checkpoint call that SinkFlowTestUtils fires even in "checkpoint disabled" tests, into TableSchema.copy()'s null-unsafe iteration in shared seatunnel-api code — and confirmed that this commit's normalizeTableSchema closes it at the single point where it can occur, with a regression test that would have caught it. I'm calling this code-complete. What I'm explicitly not claiming is that I've watched the previously-red job turn green: the Ubuntu unit-test legs that house Redis5Test/Redis7Test have now been cancelled by fail-fast on two consecutive commits, for two unrelated reasons (a Windows Maven-wrapper rate limit, then a Windows Hazelcast-node-startup flake in seatunnel-engine-server), neither of which is this PR's fault. I'd like to see that job actually complete green on this head before this merges — not because I doubt the fix, but because I said something similar two reviews ago based on an incomplete CI picture and I'd rather be precise about what I have and haven't directly observed this time. I'd suggest a job-level rerun of the cancelled unit-test (*, ubuntu-latest) legs (or the full matrix, if fail-fast keeps eating them) as the fastest path to that confirmation.
Is there a better alternative implementation? No change from my previous reviews on the overall schema-evolution approach. For this specific commit, normalizing at construction time (chosen) versus null-guarding inside TableSchema.copy() itself (the shared-code alternative I raised as a non-blocking observation in section 1.1) are both valid; the connector-local fix is the right scope for this PR, and the shared-code fix would be a reasonable, separate follow-up for whoever wants to close the same footgun for every other caller of TableSchema.copy().
DanielLeens
left a comment
DanielLeens
left a comment
There was a problem hiding this comment.
Quick follow-up: in my previous review the only remaining item was CI — specifically, I wanted to see the unit-test (8, ubuntu-latest) job (the one that houses Redis5Test/Redis7Test) actually complete green on this head, rather than relying on my hand-traced root-cause analysis alone. I just rechecked, and that exact job (and its 11, ubuntu-latest counterpart) now completed successfully on a rerun of the same run, along with the rest of the matrix. No other open source-level concerns from my side — everything flagged across my earlier rounds is fixed and re-verified. Updating my review to approve.
2 participants
[Fix][Connector-V2] Support schema evolution in Redis sink
Purpose of this pull request
Closes #11647
The Redis sink previously cached its initial
SeaTunnelRowTypeand serialization schema. Schema change events emitted by CDC sources therefore did not update Redis JSON/TEXT serialization, so newly added fields could be omitted and dropped fields could remain or become misaligned.This patch:
RedisSinkimplementSupportSchemaEvolutionSinkand declare ADD/DROP/RENAME/UPDATE COLUMN support;RedisSinkWriterimplementSupportSchemaEvolutionSinkWriter;TableSchemathroughTableSchemaChangeEventDispatcher;TableSchemawithDefaultSerializer;connector-redis-e2efor MySQL CDC -> Redis schema evolution;Does this PR introduce any user-facing change?
Yes.
Before this change, Redis values using JSON or TEXT whole-row serialization continued to use the initial source schema after a CDC DDL event. After this change, Redis serialization follows the latest physical schema for the schema change event types supported by SeaTunnel. Added fields are included, dropped fields are removed, and the latest schema is preserved across checkpoint recovery.
No connector configuration option or default value changes.
Field names configured in
key, custom key placeholders,value_field,hash_key_field, orhash_value_fieldarestatic and are not rewritten by schema evolution. Renaming or dropping one of those referenced fields while the job is
running is outside the supported scope because the existing missing-field behavior can treat the configured name as a
literal key or value.
Known scope limitation: restoring with the same parallelism, or downscaling states that carry an identical schema, is supported. Rescaling a schema-evolution job to a higher Redis sink parallelism after DDL is not addressed because SeaTunnel currently redistributes sink writer state per subtask and provides no union/broadcast writer-state mode. Supporting that case requires a separate Engine/API change rather than Redis-specific behavior.
Documentation updates:
docs/en/connectors/sink/Redis.mddocs/zh/connectors/sink/Redis.mddocs/en/introduction/configuration/schema-evolution.mddocs/zh/introduction/configuration/schema-evolution.mdHow was this patch tested?
Executed with JDK 8:
Result: full-repository formatting completed successfully.
./mvnw -pl seatunnel-connectors-v2/connector-redis \ -Dtest=RedisSinkTest,RedisSinkWriterTest testResult: 15 tests run, 0 failures, 0 errors, 0 skipped.
./mvnw -pl seatunnel-connectors-v2/connector-redis testResult: 42 tests run, 0 failures, 0 errors, 2 skipped.
Result: 7 E2E test source files compiled successfully. This command skips test execution and is reported only as E2E test compilation evidence.
Result: the Redis Connector built successfully, including Spotless checks and test compilation. Tests were skipped by this command.
Result: the full-repository build did not reach a terminal state within the 40-minute local limit and reported no failing module before timeout. This command is not reported as passed, and tests were skipped.
Result: passed after the review follow-up changes.
The new
RedisSchemaChangeITruntime could not be executed on the local machine because the Docker Desktop Linux backend was unavailable; earlier Docker diagnostics reported that firmware virtualization was disabled. The test uses the standard SeaTunnel Testcontainers E2E framework and is intended to run in CI.Check list
seatunnel-e2e/seatunnel-connector-v2-e2e/connector-redis-e2e.