[FLINK-39401] Extend raw format to support line-delimiter option

[featzhang]

Summary

This PR extends the raw format to support a new optional raw.line-delimiter config option.

When raw.line-delimiter is set:

• Deserialization: each incoming message is decoded to a string using raw.charset, split by the delimiter (String.split(Pattern.quote(delimiter), -1)), and one RowData is emitted per segment via deserialize(byte[], Collector<T>).
• Serialization: the delimiter bytes are appended to the serialized value.

When raw.line-delimiter is not set, all existing behavior is preserved exactly (backward compatible).

Example SQL

CREATE TABLE nginx_log (log STRING) WITH (
  'connector'          = 'kafka',
  'topic'              = 'nginx_log',
  'properties.bootstrap.servers' = 'localhost:9092',
  'format'             = 'raw',
  'raw.line-delimiter' = '\n'
);
-- Each Kafka message "line1\nline2\nline3" produces 3 rows
SELECT log FROM nginx_log;

Changes

Class	Change
RawFormatOptions	Add LINE_DELIMITER ConfigOption<String> with no default value
RawFormatFactory	Read new option; pass to schema builders; register in optionalOptions()
RawFormatDeserializationSchema	Override deserialize(byte[], Collector) to split by delimiter when set; add lineDelimiter field to equals/hashCode
RawFormatSerializationSchema	Append delimiter bytes when set; add lineDelimiter field to equals/hashCode
RawFormatFactoryTest	Add testLineDelimiterOption()
RawFormatLineDelimiterTest	New test class: 9 tests covering splitting, GBK charset, null message, serialization appending, null row

Test Plan

• RawFormatLineDelimiterTest (9 tests):
  • No delimiter → single row (regression)
  • \n delimiter → 3 rows
  • Custom multi-char delimiter || → 3 rows
  • Null message with delimiter → 0 rows
  • GBK charset with \n delimiter → correct splitting
  • Serialization without delimiter → no change
  • Serialization with \n → appends \n
  • Serialization with || → appends ||
  • Serialization of null row → returns null
• RawFormatFactoryTest.testLineDelimiterOption(): verifies factory produces schemas with correct delimiter
• All 50 existing raw format tests continue to pass

[flinkbot]

CI report:

• 8205f0c Azure: FAILURE

Bot commands

The @flinkbot bot supports the following commands:

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