featzhang opened a new pull request, #28048: URL: https://github.com/apache/flink/pull/28048
## What is the purpose of the change Port [PR #27897](https://github.com/apache/flink/pull/27897) (FLINK-39401) from `master` to `release-1.15`. Extends the `raw` format with an optional `raw.line-delimiter` configuration that lets each Kafka/file message encode multiple records separated by a delimiter. > Note: I'm aware `release-1.15` is past its community maintenance window. Opening this PR for visibility in case anyone else on 1.15 needs the same backport; feel free to close if out of scope. ## 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`. ## Verifying this change Added tests: - `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: ``` mvn test -pl flink-table/flink-table-runtime -Dtest='RawFormat*' ``` Result: `Tests run: 51, Failures: 0, Errors: 0, Skipped: 0` (11 new + 7 factory + 33 existing SerDe). ## 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; not porting the website docs update from #27897 since `release-1.15` docs are frozen. Happy to add if desired. -- 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]
