SEPURI-SAI-KRISHNA opened a new pull request, #13120:
URL: https://github.com/apache/seatunnel/pull/13120
Yes, glad to contribute to Apache SeaTunnel.
### Purpose of this pull request
Part of #11406, the first slice @DanielLeens proposed there: surface 1
(connector-owned state) only, no change to the bytes on disk.
When a connector's checkpoint state cannot be read on restore, for example
after an upgrade changed one of its state classes, the job fails with the bare
serializer exception and nothing says which action, job or pipeline it came
from. This PR wraps the deserialization at the four engine sites that turn
connector state bytes back into objects:
- `SourceSplitEnumeratorTask#restoreState` (enumerator state)
- `RestoredSplitOperation` (reader splits), through a new
`SourceSplitEnumeratorTask#restoreSplit`
- `SinkFlowLifeCycle#restoreState` (writer state)
- `SinkAggregatedCommitterTask#restoreState` (aggregated commit info)
A failure there now throws `StateRestoreException`, with the original
exception kept as the cause:
```
Cannot restore state of action '<action>' (job <id>, pipeline <n>): state
produced by version unknown is not readable by this connector. See the upgrade
notes for <connector>. Cause: <original exception>
```
- The version is "unknown" and the readable range is left out until the
version metadata from the STIP exists.
- The checkpoint id is not in the message yet. It does not reach these call
sites, and how to add it is the open question on #11406. The coordinator
already logs it on restore (`restore with SAVEPOINT on checkpointId(4)` in the
run below).
- Zeta wraps every sink in `MultiTableSink`, so for sinks the connector name
is taken from the wrapped sinks (`LocalFile`, not `MultiTableSink`).
- Each site changes by one call, which keeps the overlap with the restore
paths in #11557 small.
### Does this PR introduce _any_ user-facing change?
Only the error when connector state cannot be restored. The job still fails
at the same point, and the original exception is still in the cause chain. No
option, state format or interface changes.
### How was this patch tested?
**Unit tests** (JDK 17): `StateRestoreExceptionTest` (new, for the new
class) and one case each in `SourceSplitEnumeratorTaskTest` (two: enumerator
state and `restoreSplit`), `SinkAggregatedCommitterTaskTest` and
`SinkFlowLifeCycleErrorOutcomeTest`. The fixture is real: state bytes written
by a class with `serialVersionUID` 1, read by the same class with 2, which
throws the `InvalidClassException` an upgrade produces. Putting the `dev`
deserialization back in any one of the three task classes fails exactly the new
tests for that class. The `checkpoint` and `task` test packages pass (198 run,
0 failures; the 2 skipped are `@Disabled` on `dev`).
**Real restore on Zeta**, two JVMs, like an upgrade: run
`stream_fakesource_to_file_savepoint.conf` (FakeSource to LocalFile) on `dev`
and take a savepoint, then restore it in a new JVM with one connector state
class given a new `serialVersionUID`. Job result on `dev` with `FileSinkState`
changed:
```
org.apache.seatunnel.engine.server.checkpoint.CheckpointException:
CheckpointCoordinator inside have error.
Caused by: org.apache.seatunnel.common.utils.SeaTunnelException:
org.apache.seatunnel.common.utils.SerializationException:
java.io.InvalidClassException:
org.apache.seatunnel.connectors.seatunnel.file.sink.state.FileSinkState; local
class incompatible: stream classdesc serialVersionUID = -8757454855081836294,
local class serialVersionUID = 1
```
With this PR:
```
org.apache.seatunnel.engine.server.checkpoint.CheckpointException:
CheckpointCoordinator inside have error.
Caused by: org.apache.seatunnel.common.utils.SeaTunnelException:
org.apache.seatunnel.engine.server.checkpoint.StateRestoreException: Cannot
restore state of action 'pipeline-1 [Sink[0]-LocalFile-MultiTableSink]' (job
823342777, pipeline 1): state produced by version unknown is not readable by
this connector. See the upgrade notes for LocalFile. Cause:
org.apache.seatunnel.common.utils.SerializationException:
java.io.InvalidClassException:
org.apache.seatunnel.connectors.seatunnel.file.sink.state.FileSinkState; local
class incompatible: stream classdesc serialVersionUID = -8757454855081836294,
local class serialVersionUID = 1
```
The same before/after holds with `FakeSourceSplit` changed (enumerator site,
"FakeSource") and with `FileAggregatedCommitInfo` changed (aggregated committer
site, "LocalFile"). Restoring with no class changed reaches RUNNING from
checkpoint 4 on both `dev` and this PR. The reader split site is covered by the
unit test only: with FakeSource the enumerator state holds the same split
class, so the enumerator fails first and the pending `RestoredSplitOperation`
is cancelled.
Also run: `spotless:check`.
### Check list
* [x] If any new Jar binary package adding in your PR, please add License
Notice according
[New License
Guide](https://github.com/apache/seatunnel/blob/dev/docs/en/developer/new-license.md)
(no new jar)
* [x] If necessary, please update the documentation to describe the new
feature. https://github.com/apache/seatunnel/tree/dev/docs (not needed, error
message only)
* [x] If necessary, please update `incompatible-changes.md` to describe the
incompatibility caused by this PR. (not needed)
--
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]
