Flink cdc3.0动态变更表结构——源码解析 - 服务器托管|北京服务器租用|机房托管租用|IDC托管租用|机房机柜带宽租用-价格及费用咨询

文章目录

前言
源码解析
- 1. 接收schema变更事件
- 2. 发起schema变更请求
- 3. schema变更请求具体处理
- 4. 广播刷新事件并阻塞
- 5. 处理FlushEvent
- 6. 修改sink端schema
结尾

前言

上一篇Flink cdc3.0同步实例介绍了最新的一些功能和问题，本篇来看下新功能之一的动态变更表结构的具体实现。
在 Flink 中，应用程序由流数据流组成，这些数据流是由用户定义的Operators进行转换。

Flink CDC 3.0 框架中流动的数据类型被称为Event，代表外部系统产生的变更事件。每个事件都标有发生更改的表 ID 。事件分为SchemaChangeEvent和DataChangeEvent，分别代表表结构和数据的变化。处理schema变更的Operators对应图中的SchemaOperator。

（以下代码使用Flink Release 3.0.0）

源码解析

1. 接收schema变更事件

我们以添加字段触发的AddColumnEvent为例，它实现了SchemaChangeEvent。 SchemaOperator 当接收到有AddColumnEvent 事件时，会在processElement 中调用handleSchemaChangeEvent处理。

2. 发起schema变更请求

说明下这里的response实际是直接返回的new SchemaChangeResponse(true)，由于构造的shouldSendFlushEvent 直接传入true，所以后续也会进入if条件。我们接着requestSchemaChange 方法看

由于知道response是直接创建的已知结果，因此responseFuture.get() 也不会阻塞。我们接着来看toCoordinator.sendRequestToCoordinator(getOperatorID(), new SerializedValue(request));的实现

3. schema变更请求具体处理

通过几层的调用，上述变更请求会走到 SchemaRegistry 的handleCoordinationRequest(CoordinationRequest request)，我们的请求是SchemaChangeRequest，所以会调用requestHandler.handleSchemaChangeRequest(schemaChangeRequest);

这里可以看到response 是直接创建的SchemaChangeResponse(true)。接着schemaManager.applySchemaChange(request.getSchemaChangeEvent());注册新的schema。

另外还有个重点，在startToWaitForReleaseRequest方法中会重置responseFuture，原本的response通过return返回了。而PendingSchemaChange中的response重置，主要就是为了等schema变更完成设计。（主线程会再次发起请求调用responseFuture.get() ，忽略这里会不理解后面为什么会阻塞）

4. 广播刷新事件并阻塞

回到第二部分，因为response是一个明确对象没有阻塞，返回后会直接广播FlushEvent 和 schemaChangeEvent服务器托管网（再次发起schemaChangeEvent不是很理解）。之后requestReleaseUpstream 请求调用responseFuture.get()会阻塞，因为response在第三步已经重置为new CompletableFuture()，利用的1.8的特性。这也是收到变更事件后要保证sink端变更才能发放数据。

5. 处理FlushEvent

FlushEvent 由什么Operator处理，在官方架构图中其实没有指出，但是图标可以看出是通过sink端完成，我们可以找到DataSin服务器托管网kWriterOperator类，有对FlushEvent的处理。

实际调用SchemaRegistry::handleEventFromOperator方法，重点在requestHandler.flushSuccess(flushSuccessEvent.getTableId(), flushSuccessEvent.getSubtask());

其中applySchemaChange 就是在具体的sink端变更，下面会展开。当变更完成后会执行waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));，实际就通知第4部分的response这里处理完了，可以正常放开数据流。