如何做网站聚合页,旅游网站建设实训报告,开发公司销售人员竞聘演讲稿,个人logo在线生成免费文章目录 前言源码解析1. 接收schema变更事件2. 发起schema变更请求3. schema变更请求具体处理4. 广播刷新事件并阻塞5. 处理FlushEvent6. 修改sink端schema 结尾 前言
上一篇Flink cdc3.0同步实例 介绍了最新的一些功能和问题#xff0c;本篇来看下新功能之一的动态变更表结… 文章目录 前言源码解析1. 接收schema变更事件2. 发起schema变更请求3. schema变更请求具体处理4. 广播刷新事件并阻塞5. 处理FlushEvent6. 修改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端完成我们可以找到DataSinkWriterOperator类有对FlushEvent的处理。 实际调用SchemaRegistry::handleEventFromOperator方法重点在requestHandler.flushSuccess(flushSuccessEvent.getTableId(), flushSuccessEvent.getSubtask()); 其中applySchemaChange 就是在具体的sink端变更下面会展开。 当变更完成后会执行waitFlushSuccess.getResponseFuture().complete(wrap(new ReleaseUpstreamResponse()));实际就通知第4部分的response这里处理完了可以正常放开数据流。
6. 修改sink端schema
每个sink端有自定义的metadataApplier 我们以DorisMetadataApplier为例applyAddColumnEvent 会构造addFieldSchema然后在schemaChangeManager 中转换为对应的sql执行。
结尾
以上就是这两天对源码跟进的记录后续思考使用local环境Debug中间过程。
