featzhang opened a new pull request, #28082: URL: https://github.com/apache/flink/pull/28082
## What is the purpose of the change Port [PR #27897](https://github.com/apache/flink/pull/27897) (FLINK-39401) from `master` to `release-2.1`. Extends the `raw` format with an optional `raw.line-delimiter` configuration that lets each Kafka/file message encode multiple records separated by a delimiter. This is a clean cherry-pick with **zero conflicts** — the raw format code on `release-2.1` has not diverged from `master` for the touched classes, so the backport is byte-for-byte identical to the master version (file content + tests + docs). ## Brief change log - `RawFormatOptions`: add `LINE_DELIMITER` `ConfigOption` (no default, supports Java escape sequences like `\n`, `\r\n`). - `RawFormatFactory`: read the option, register it in `optionalOptions()`, and pass it to the (de)serialization schemas. - `RawFormatDeserializationSchema`: - new 5-arg constructor accepting `@Nullable String lineDelimiter`; the previous 4-arg constructor delegates with `null` for backward compatibility. - pre-compiled `Pattern` for splitting; new `deserialize(byte[], Collector<RowData>)` override emits one `RowData` per segment. - null message with delimiter → zero rows; trailing delimiter stripped so round-trip produces one row, not two. - `RawFormatSerializationSchema`: - new 4-arg constructor accepting `@Nullable String lineDelimiter`; old 3-arg constructor delegates with `null`. - pre-computed `delimiterBytes`; `serialize()` appends them after the value bytes. `null` row still returns `null`. - Docs: `docs/content/docs/connectors/table/formats/raw.md` and the Chinese mirror are updated with the new option row. ## Verifying this change Added tests (unchanged from #27897): - `RawFormatFactoryTest.testLineDelimiterOption` — verifies the factory wires the option through correctly. - `RawFormatLineDelimiterTest` (new, 11 tests) — covers: - deserialize without / with `\n` / with multi-char / with GBK charset delimiters - null message, trailing delimiter, round-trip - serialize without / with `\n` / with custom delimiter, null row Run locally against `release-2.1`: ``` ./mvnw test -pl flink-table/flink-table-runtime -Dtest='RawFormat*' ``` Result: `Tests run: 52, Failures: 0, Errors: 0, Skipped: 0` (11 new + 7 factory + 34 existing SerDe). All pre-existing tests continue to pass. Also ran `./mvnw spotless:check -pl flink-table/flink-table-runtime` → clean. ## Does this pull request potentially affect one of the following parts: - Dependencies (does it add or upgrade a dependency): **no** - The public API, i.e., is any changed class annotated with `@Public(Evolving)`: **no** (new option is additive, behavior unchanged when unset) - The serializers: **no** - The runtime per-record code paths (performance sensitive): **yes** — raw format `deserialize`/`serialize`. When the option is unset, the behavior and allocations are unchanged; when set, a pre-compiled `Pattern` and pre-computed `byte[]` avoid per-record allocation. - Anything that affects deployment or recovery: JobManager (and its components), Checkpointing, Kubernetes/Yarn, ZooKeeper: **no** - The S3 file system connector: **no** ## Documentation - Does this pull request introduce a new feature? **yes** - If yes, how is the feature documented? JavaDocs on the `LINE_DELIMITER` option + website docs (`docs/content{,.zh}/docs/connectors/table/formats/raw.md`), both ported from #27897. -- 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]
