[FLINK-35104] [pipeline-connector/kafka]support pipeline kafka source

[MOBIN-F]

support pipeline kafka source

Core Point:

1.

processDeferredRequests() primarily addresses the scenario where a SchemaCoordinator is in the middle of executing one round of schema evolution when it receives another SchemaChangeRequest. This is quite common in the scenario of Kafka's multi-partition setup.

This processDeferredRequests() mainly resolves the issue: when a SchemaCoordinator is executing a round of schema evolution, it receives another SchemaChangeRequest.

Typical Scenario

Assume the Schema Operator has a parallelism of 2 and consecutively receives three changes:

1. CREATE TABLE users
2. ADD COLUMN users.name
3. ADD COLUMN users.email

The first round of CREATE TABLE has collected requests from both subtasks:

subtask-0 -> CREATE TABLE
subtask-1 -> CREATE TABLE

Coordinator status changed to:

IDLE -> WAITING_FOR_FLUSH -> EVOLVING

MetadataApplier is executing CREATE TABLE in an external system. If an ADD COLUMN name request arrives at this moment, the code cannot process it immediately because:

• The external table may not yet be fully created;
• The evolved schema has not been updated;
• The two changes cannot share the same flush barrier;
• If executed in parallel, ADD COLUMN might occur before CREATE TABLE.

Therefore, the request is placed into a deferred queue:

if (evolvingStatus.get() == RequestStatus.EVOLVING) {
LOG.info(
"Schema evolution is in progress. Deferring request {} to the next round.",
request);
deferredRequests.add(Tuple2.of(request, responseFuture));
return;
}

After the first round is completed, call:

processDeferredRequests();

Reassign ADD COLUMN name to the state machine to start the next round:

Round 1: CREATE TABLE users  
↓ Completed  
Round 2: ADD COLUMN users.name  
↓ Completed  
Round 3: ADD COLUMN users.email

2.

Kafka is declared as a parallel metadata source, so it follows a distributed topology. Schema events are broadcast to all SchemaOperators.
However, the following situation may occur, pipelin kafka source ---> paimon:
Suppose the parallelism is 2, and at the same time receive:

• Kafka partition 0: CREATE customers
• Kafka partition 1: CREATE orders

Cross-processing may occur:

• SchemaOperator-0 processes source 1 first
• SchemaOperator-1 processes source 0 first

At this time:

• BucketAssign-0 only receives the Flush of source 1
• BucketAssign-1 only receives the Flush of source 0

And the FlushEventAlignmentOperator requires that the same source must receive all the reports from BucketAssign. Therefore, both sides will keep waiting.

Solution: Add a FlushReplicateOperator before the BucketAssignOperator at the sink end of the Kafka distributed topology, so that the Flush is replicated to all BucketAssign.

[MOBIN-F]

According to the official documentation of Starrocks, modifying column types is asynchronous. You need to wait until Starrocks has completely completed the change before proceeding. Otherwise, an exception will occur.
For example, if changing from int to string, if using executeUpdateStatement, it will return immediately. However, the actual Starrocks table is still of int type, resulting in the failure of inserting string-type data.

Operations on bucket, column and rollup are asynchronous operations. A success message is return immediately after the task is submitted. You can run the SHOW ALTER TABLE command to check the progress, and run the CANCEL ALTER TABLE command to cancel the operation.