[FLINK-39401][formats] Port raw line-delimiter option to release-1.15

[featzhang]

What is the purpose of the change

Port PR #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 [FLINK-39401] Extend raw format to support line-delimiter option #27897 since release-1.15 docs are frozen. Happy to add if desired.

[flinkbot]

CI report:

• d5efa5f Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

• @flinkbot run azure re-run the last Azure build

[spuru9]

@featzhang flinkbot is acting weirdly. Try making a empty comment. git commit -m "trigger" --allow-empty.