1. 为什么非得自己写一个 Flink SQL Connector我第一次在某跨平台实时数仓项目里遇到这个问题是在凌晨两点——上游业务方突然要求把一份存放在自研对象存储里的 JSON 日志以“表”的形式直接接入 Flink SQL 作业延迟要压到 500ms 以内。当时团队里所有人都下意识打开官网文档翻到FileSystemconnector试了十分钟就放弃了它只支持s3://、hdfs://、file://这几种协议前缀而我们那个对象存储用的是ossx://一个完全自定义的协议连FileSystemFactory都没注册过。更麻烦的是它的 schema 推断机制对嵌套 JSON 的处理是硬编码的而我们的日志字段会动态增减必须支持ROW...类型的运行时解析。这时候才真正意识到Flink SQL 看似开箱即用但它的 connector 生态本质是个“协议适配层”不是万能胶水。官方提供的kafka、jdbc、filesystem、pulsar等 connector背后全是标准协议 固定元数据模型 预设序列化策略的组合。一旦你的数据源不走标准协议比如用私有 RPC 调用拉取、不遵循通用 schema 规范比如字段名带点号、类型随时间漂移、或者需要特殊权限控制比如每个租户 token 不同那CREATE TABLE语句里写的WITH参数就立刻失效——你不是在配置 connector而是在和它的抽象边界打架。这其实暴露了一个被很多人忽略的事实Flink SQL 的执行引擎Planner和 connector 是解耦的但解耦不等于无感。SQL 层负责逻辑计划生成connector 负责物理计划落地。当你写SELECT * FROM user_logFlink 不会去猜user_log表的数据从哪来、怎么分片、如何 checkpoint、失败后怎么重试它只认 connector 实现的DynamicTableSource接口。这个接口里藏着五个关键契约如何发现分区Split Discovery、如何读取数据Scan Runtime、如何描述表结构Schema、如何处理水印Watermark Strategy、如何恢复状态Restore Logic。任何一个环节没对齐SQL 就跑不起来或者跑起来就丢数、重复、乱序。所以“自定义 connector”从来不是炫技而是当你的数据源长成这样时的唯一出路数据源没有标准 JDBC 驱动但提供了 HTTP API 和 OAuth2 Token数据格式不是纯 JSON 或 CSV而是 Protobuf 编码 自定义压缩头分区不是按时间目录划分而是按业务 ID 哈希后落在不同微服务实例上每次读取都要携带租户上下文Header 中传 X-Tenant-ID且 token 有效期只有 15 分钟需自动刷新数据本身不带事件时间戳但业务规则要求按“日志生成时间”做窗口计算这个时间藏在消息体 Base64 解码后的某个字节偏移量里。这些需求你翻遍 Flink 1.17 的所有内置 connector 源码都找不到现成实现。它们不是 bug而是设计边界——Flink 社区只保证通用性不承诺全覆盖。而生产环境里90% 的“非标数据源”恰恰卡在这条边界上。所以这篇内容不讲“怎么搭个 demo”而是带你从零抠出一个真实可用的 connector它能跑通CREATE TABLE、支持SELECT流式查询、兼容INSERT INTO写入、通过CHECKPOINT保证 exactly-once、还能在 Web UI 上看到真实的 source parallelism 和 lag 指标。所有代码基于 Flink 1.18所有路径可直接复制粘贴进你的pom.xml。提示别被“自定义”吓住。Flink 官方把 connector 抽象成三层API 层用户写的DynamicTableSource、Runtime 层Flink 框架调用的InputFormat或SourceFunction、IO 层你封装的 HTTP Client / Kafka Consumer。我们只聚焦最上面一层下面两层用 Flink 已验证的基类兜底工作量比你想象中少得多。2. 从零拆解一个可上线的 connector 必须包含哪五个核心组件写一个能进生产环境的 connector不是堆砌代码而是构建一套可验证、可监控、可运维的契约实现。我把它拆成五个不可妥协的核心组件每个组件对应 Flink SQL 运行时的一个关键决策点。漏掉任何一个你的 connector 在压测或上线后必然暴雷。2.1 Schema 解析器让 Flink “看懂”你的数据结构Flink SQL 的CREATE TABLE语句里写的FIELDS最终要映射成 Java 的RowType或DataType。但很多自研数据源根本不提供 schema 元数据接口——比如一个 HTTP 日志服务你只能拿到 raw bytes字段定义全靠文档约定。这时候DeserializationSchema就成了第一道生死线。我见过太多人在这里踩坑直接用SimpleStringSchema把整行当字符串然后在 SQL 里用JSON_VALUE()函数解析结果吞吐量暴跌 70%。正确做法是在 connector 初始化阶段就完成 schema 的静态绑定或动态推断。以我们实际项目为例数据源返回的 JSON 样例是{ event_id: evt_abc123, user: { id: 1001, name: 张三, tags: [vip, active] }, ts_ms: 1712345678901 }如果硬编码ROWevent_id STRING, user ROWid BIGINT, name STRING, tags ARRAYSTRING, ts_ms BIGINT后续字段变更就得发版重启。所以我们做了两件事定义SchemaProviderSPI 接口允许外部注入 schema 获取逻辑比如从 Consul 读取最新版本或调用/schema/v1HTTP 接口实现JsonRowDataDeserializationSchema继承AbstractDeserializationSchemaRowData重写deserialize(byte[] message)方法在反序列化时动态解析 JSON 结构用GenericRowData构建RowData实例并缓存TypeInformation避免重复反射。关键细节在于isEndOfStream()的实现——很多 connector 直接return false导致 Flink 认为流永远不结束checkpoint 无法触发。我们必须根据业务语义判断比如当 HTTP 响应头X-Has-More: false时才返回true。2.2 Split 划分器告诉 Flink “数据在哪、怎么分”Flink 的并行读取依赖Split模型。内置FileSystemconnector 会把每个文件当作一个FileSplitKafkaconnector 把每个 partition 当作KafkaPartitionSplit。但如果你的数据源是 HTTP API它没有“分区”概念只有“页码”或“游标”。这时Split就是你定义并行度的画布。我们设计了一个HttpApiSplit类包含三个字段cursor: String当前页游标如next_tokenabc123tenantId: String租户隔离标识retryCount: int失败重试次数用于幂等控制初始化时SplitEnumerator调用listSplits()方法向/api/v1/logs/splits发起请求返回 8 个HttpApiSplit实例对应 8 并行度。每个HttpSourceReader拿到一个 split 后就循环调用/api/v1/logs?cursor{split.cursor}tenant{split.tenantId}直到响应为空或超时。这里有个致命陷阱split 必须可序列化且幂等。早期我们把OkHttpClient实例塞进HttpApiSplit结果 checkpoint 失败——因为OkHttpClient不是Serializable。后来改成只存原始参数SourceReader在open()时再创建 client。另外cursor必须是字符串不能是MapString, Object否则 Flink 的 state backend 序列化会报错。2.3 Source Function把 Split 转成真正的数据流DynamicTableSource只是声明“我能读”真正干活的是ScanRuntimeProvider返回的InputFormat或SourceFunction。Flink 1.17 推荐用SourceAPI即SourceFunction的新形态因为它原生支持 watermark、checkpoint、多并行度。我们继承RichSourceFunctionRowData核心逻辑在run(SourceContextRowData ctx)方法里从getRuntimeContext().getIndexOfThisSubtask()拿到当前 subtask ID从state或configuration中获取分配给它的HttpApiSplit启动一个 while 循环每次请求后调用ctx.collectWithTimestamp(rowData, eventTime)注入事件时间在循环内捕获IOException若重试超过split.retryCount则调用ctx.markAsTemporarilyIdle()让 Flink 暂停该 subtask避免拖垮整个 job。重点来了ctx.collectWithTimestamp()的第二个参数eventTime必须是毫秒级 long且不能为负数。我们从 JSON 里解析ts_ms字段但如果字段缺失就 fallback 到系统当前时间System.currentTimeMillis()。这个 fallback 看似合理实则埋雷——当大量数据缺失时间戳时watermark 会卡死在当前时间下游窗口永远不触发。最终方案是定义一个EventTimeExtractorSPI允许配置 fallback 策略如LATEST_SEEN或NEVER_FALLBACK并在open()时校验ts_ms是否存在不存在则抛异常中断启动强制业务方补数据。2.4 Watermark 策略解决“数据迟到”这个永恒难题Flink 的 event-time 处理依赖 watermark。内置 connector 通常提供WATERMARK FOR ts AS ...语法但底层必须由 connector 实现WatermarkStrategy。很多人以为只要 SQL 里写了WATERMARK FOR ts AS ts - INTERVAL 5 SECONDFlink 就会自动处理其实不然——这个语法只是告诉 planner “请用这个字段生成 watermark”真正的生成逻辑在Source的getWatermarkStrategy()方法里。我们实现了一个HttpApiWatermarkStrategy继承WatermarkStrategyRowDatacreateWatermark()方法从RowData里提取ts_ms字段减去 5 秒返回Watermark对象onPeriodicEmit()方法每 200ms 调用一次确保 watermark 持续推进关键是onEvent()方法当收到一条新数据时更新当前最大ts_ms并检查是否需要立即发射新 watermark比如数据乱序严重时。这里有个反直觉的细节onPeriodicEmit()的周期不是越小越好。我们实测过设成 50ms 后jobManager CPU 占用飙升 40%因为 watermark 发射太频繁触发了过多的窗口计算。最终定为 200ms既保证低延迟又不增加调度压力。2.5 State 恢复逻辑让故障重启不丢数、不重复Exactly-once 的灵魂是 state。Flink 的SourceFunction必须实现CheckpointedFunction接口把split、cursor、maxSeenTimestamp等运行时状态持久化。我们定义了一个HttpApiSourceStatePOJOpublic class HttpApiSourceState implements Serializable { private String cursor; private long maxSeenTimestamp; private int retryCount; private MapString, Object metadata; // 存储 HTTP 响应头等上下文 }在snapshotState(FunctionSnapshotContext context)里把当前split的状态序列化进context在restoreState(Collection state)里反序列化并恢复cursor。但这里有个大坑restoreState可能在 job 启动时被多次调用。我们曾遇到过Flink 在恢复时先调用一次restoreState加载旧状态紧接着又调用一次加载新 checkpoint导致cursor被覆盖回旧值数据重放。解决方案是在restoreState里加锁只取 collection 中maxSeenTimestamp最大的那个 state。注意不要在snapshotState里做网络 I/O曾经有同事为了“保险”在 checkpoint 时调用/api/v1/health检查服务状态结果 checkpoint 超时失败。记住checkpoint 是纯内存操作所有外部依赖必须剥离。3. 实战手把手从 Maven 依赖到第一个可运行的 connector现在把前面五个组件串起来写一个最小但可运行的 connector。我们以“从 HTTP API 读取 JSON 日志”为场景目标是让这条 SQL 能成功执行CREATE TABLE http_logs ( event_id STRING, user_id BIGINT, user_name STRING, ts_ms BIGINT, WATERMARK FOR ts_ms AS ts_ms - INTERVAL 5 SECOND ) WITH ( connector http-api, url https://api.example.com/v1/logs, tenant-id prod-001, token xxx, scan.interval 10000 );3.1 Maven 依赖精简到只留必要项Flink connector 开发最怕依赖冲突。Flink 自身用的是slf4j-api1.7.x但很多 HTTP client 依赖 2.0直接引入会导致NoClassDefFoundError。我们的pom.xml只保留四组依赖dependencies !-- Flink 核心 API必须与集群版本严格一致 -- dependency groupIdorg.apache.flink/groupId artifactIdflink-table-api-java/artifactId version1.18.0/version scopeprovided/scope /dependency dependency groupIdorg.apache.flink/groupId artifactIdflink-connector-base/artifactId version1.18.0/version /dependency !-- HTTP 客户端选 OkHttp 是因为它的 ConnectionPool 天然支持连接复用 -- dependency groupIdcom.squareup.okhttp3/groupId artifactIdokhttp/artifactId version4.12.0/version /dependency !-- JSON 解析用 Jackson 是因为 Flink 内部也用它避免多套解析器 -- dependency groupIdcom.fasterxml.jackson.core/groupId artifactIdjackson-databind/artifactId version2.15.2/version /dependency /dependencies关键点所有flink-*依赖 scope 设为provided防止打包进 fat jar 后与集群冲突okhttp和jackson不设provided因为它们是 connector 私有依赖不会和 Flink 冲突版本号必须精确匹配比如 Flink 1.18.0 对应flink-connector-base1.18.0差一个小版本都可能编译失败。3.2 Connector Factory让 Flink “认识”你的 connector这是CREATE TABLE语句里connector http-api背后的注册点。必须实现DynamicTableFactory接口并添加AutoService注解让 SPI 机制自动发现AutoService(DynamicTableFactory.class) public class HttpApiDynamicTableFactory implements DynamicTableFactory { Override public String factoryIdentifier() { return http-api; // 必须和 SQL 里写的完全一致 } Override public DynamicTableSource createDynamicTableSource(Context context) { final FactoryUtil.TableFactoryHelper helper FactoryUtil.createTableFactoryHelper(this, context); helper.validate(); final ReadableConfig options helper.getOptions(); final String url options.get(UrlOption.URL); final String tenantId options.get(TenantOption.TENANT_ID); final String token options.get(TokenOption.TOKEN); // 构建 Schema final DataType dataType context.getPhysicalRowDataType(); final DeserializationSchemaRowData deserializer new JsonRowDataDeserializationSchema(dataType, true, true); // 构建 Source final HttpApiSource source new HttpApiSource( url, tenantId, token, options.get(ScanIntervalOption.SCAN_INTERVAL), deserializer ); return new HttpApiDynamicTableSource(source, dataType); } }注意factoryIdentifier()返回的字符串必须小写、无下划线、无空格且全局唯一。如果写成http_apiSQL 里就必须写connector http_api否则 Flink 找不到 factory。3.3 DynamicTableSourceSQL 层和 Runtime 层的桥梁这个类是 connector 的门面它把Source包装成 Flink SQL 能理解的DynamicTableSourcepublic class HttpApiDynamicTableSource implements DynamicTableSource { private final HttpApiSource source; private final DataType producedDataType; public HttpApiDynamicTableSource(HttpApiSource source, DataType producedDataType) { this.source source; this.producedDataType producedDataType; } Override public ScanRuntimeProvider getScanRuntimeProvider(ScanContext runtimeContext) { return SourceFunctionProvider.of( source, true // 是否支持并行 ); } Override public ChangelogMode getChangelogMode() { return ChangelogMode.insertOnly(); // 我们的日志是 append-only } Override public DynamicTableSource copy() { return new HttpApiDynamicTableSource(source, producedDataType); } Override public String asSummaryString() { return HttpApiDynamicTableSource; } }最关键的getScanRuntimeProvider()方法它返回SourceFunctionProvider把HttpApiSource我们写的SourceFunction交给 Flink 运行时。ChangelogMode.insertOnly()告诉 Flink 这是追加流不需要处理UPDATE_BEFORE/UPDATE_AFTER。3.4 HttpApiSource真正的数据搬运工这是整个 connector 的心脏。我们继承RichSourceFunctionRowData并实现CheckpointedFunctionpublic class HttpApiSource extends RichSourceFunctionRowData implements CheckpointedFunction { private final String url; private final String tenantId; private final String token; private final long scanIntervalMs; private final DeserializationSchemaRowData deserializer; // 运行时状态 private volatile boolean isRunning true; private transient OkHttpClient httpClient; private transient HttpApiSplit currentSplit; // checkpoint 状态 private HttpApiSourceState state; public HttpApiSource(String url, String tenantId, String token, long scanIntervalMs, DeserializationSchemaRowData deserializer) { this.url url; this.tenantId tenantId; this.token token; this.scanIntervalMs scanIntervalMs; this.deserializer deserializer; } Override public void open(Configuration parameters) throws Exception { super.open(parameters); // 初始化 HTTP 客户端 this.httpClient new OkHttpClient.Builder() .connectTimeout(10, TimeUnit.SECONDS) .readTimeout(30, TimeUnit.SECONDS) .build(); // 初始化 split这里简化为单 split实际应从 SplitEnumerator 获取 this.currentSplit new HttpApiSplit(initial_cursor, tenantId, 3); this.state new HttpApiSourceState(); } Override public void run(SourceContextRowData ctx) throws Exception { while (isRunning) { try { // 构造请求 Request request new Request.Builder() .url(url ?cursor currentSplit.getCursor() tenant tenantId) .header(Authorization, Bearer token) .build(); Response response httpClient.newCall(request).execute(); if (!response.isSuccessful()) { throw new IOException(HTTP response.code() : response.message()); } // 解析响应体 String body response.body().string(); JSONArray jsonArray new JSONArray(body); for (int i 0; i jsonArray.length(); i) { JSONObject json jsonArray.getJSONObject(i); RowData rowData deserializer.deserialize(json.toString().getBytes(StandardCharsets.UTF_8)); long eventTime extractEventTime(rowData); // 从 rowData 提取 ts_ms ctx.collectWithTimestamp(rowData, eventTime); } // 更新 cursor String nextCursor response.header(X-Next-Cursor, ); if (nextCursor.isEmpty()) { Thread.sleep(scanIntervalMs); // 没有下一页休眠 } else { currentSplit new HttpApiSplit(nextCursor, tenantId, currentSplit.getRetryCount()); } } catch (Exception e) { LOG.error(Error reading from HTTP API, e); Thread.sleep(1000); // 出错后降频重试 } } } Override public void cancel() { isRunning false; } Override public void snapshotState(FunctionSnapshotContext context) throws Exception { state.setCursor(currentSplit.getCursor()); state.setMaxSeenTimestamp(System.currentTimeMillis()); // 将 state 序列化到 checkpoint context.getCheckpointOutputView().writeUTF(JsonUtils.toJson(state)); } Override public void restoreState(FunctionSnapshotContext context) throws Exception { String json context.getCheckpointInputView().readUTF(); this.state JsonUtils.fromJson(json, HttpApiSourceState.class); this.currentSplit new HttpApiSplit( state.getCursor(), tenantId, 3 ); } }这段代码已经能跑通基础功能。但要注意deserializer.deserialize()输入是byte[]所以要把JSONObject.toString()转成 UTF-8 bytesctx.collectWithTimestamp()必须在while循环内调用不能在run()外围调用snapshotState和restoreState的序列化格式必须一致我们用JsonUtilsJackson 封装保证兼容性。3.5 打包与部署三步让 connector 进入 Flink 集群写完代码不能直接扔进lib/目录——Flink 1.15 默认禁用用户 JAR 的反射必须显式注册。分三步打包成 uber-jar用 Maven Shade Plugin 打包排除slf4j-api、commons-logging等冲突包plugin groupIdorg.apache.maven.plugins/groupId artifactIdmaven-shade-plugin/artifactId version3.4.1/version executions execution phasepackage/phase goalsgoalshade/goal/goals configuration filters filter artifact*:*/artifact excludes excludeorg/slf4j/**/exclude excludecommons-logging/**/exclude /excludes /filter /filters /configuration /execution /executions /plugin拷贝到 Flink lib 目录cp target/http-api-connector-1.0.jar $FLINK_HOME/lib/重启 Flink Session Cluster$FLINK_HOME/bin/start-cluster.sh。注意Per-Job Cluster 模式下需用-C参数指定 JARflink run -C file:///path/to/http-api-connector-1.0.jar ...提示首次部署后登录 Flink Web UI在Configuration页面搜索table.factories确认http-api出现在列表中。如果没出现说明AutoService注解没生效检查META-INF/services/org.apache.flink.table.factories.DynamicTableFactory文件是否生成。4. 生产级加固监控、容错、性能调优的六个实战技巧写一个能跑的 connector 只完成了 30%剩下 70% 是让它在生产环境稳如磐石。以下是我在多个项目中沉淀下来的六个关键技巧每个都来自真实故障现场。4.1 指标埋点让 connector “会说话”Flink 的SourceFunction可以注册指标Metrics但默认不开启。我们在open()方法里主动注册Override public void open(Configuration parameters) throws Exception { super.open(parameters); MetricGroup metricGroup getRuntimeContext().getMetricGroup(); this.httpRequestCounter metricGroup.counter(http_requests_total); this.httpErrorCounter metricGroup.counter(http_errors_total); this.bytesReadGauge metricGroup.gauge(bytes_read, () - totalBytesRead); this.lagGauge metricGroup.gauge(data_lag_ms, () - System.currentTimeMillis() - lastEventTime); }然后在run()方法里打点// 请求前 httpRequestCounter.inc(); // 解析成功后 totalBytesRead body.length(); // 出错时 httpErrorCounter.inc();部署后在 Web UI 的Task Managers→Metrics页面就能看到这些指标。我们曾靠data_lag_ms发现某租户的 token 过期导致lag持续增长到 5 分钟而其他指标一切正常。4.2 Token 自动刷新解决 15 分钟过期魔咒HTTP API 的 token 有效期短是常态。手动轮换不现实必须自动。我们在HttpApiSource里加了一个TokenRefresher线程private void startTokenRefresher() { new Thread(() - { while (isRunning) { try { // 提前 30 秒刷新 if (System.currentTimeMillis() - lastRefreshTime 870_000) { String newToken refreshTokenFromAuthServer(); this.token newToken; lastRefreshTime System.currentTimeMillis(); } Thread.sleep(30_000); } catch (Exception e) { LOG.warn(Token refresh failed, e); } } }).start(); }关键点refreshTokenFromAuthServer()必须是幂等的且刷新过程不能阻塞主数据流。我们把 token 存在volatile String token字段里所有 HTTP 请求都读这个字段保证原子性。4.3 连接池调优把 QPS 从 200 干到 2000OkHttp 的ConnectionPool默认最大空闲连接 5 个保活 5 分钟。对于高并发 HTTP 读取这远远不够。我们在open()里定制this.httpClient new OkHttpClient.Builder() .connectionPool(new ConnectionPool(20, 5, TimeUnit.MINUTES)) // 20 个空闲连接 .build();同时scan.interval不能设得太小。我们测试发现当scan.interval100ms时QPS 达到 2000但错误率飙升到 15%连接池耗尽设为500ms后QPS 稳定在 1800错误率低于 0.1%。所以最终配置是scan.interval500ConnectionPool(20)。4.4 内存泄漏防护避免 OOM 杀死 TaskManagerJSON 解析容易引发内存泄漏。JSONArray和JSONObject是 Jackson 的TreeModel会缓存整个 JSON 树。我们改用JsonParser流式解析JsonFactory factory new JsonFactory(); try (JsonParser parser factory.createParser(jsonString)) { while (parser.nextToken() ! JsonToken.END_ARRAY) { if (parser.getCurrentToken() JsonToken.START_OBJECT) { // 逐个字段解析不构建完整对象 String eventId parser.nextTextValue(); // event_id 字段 parser.nextToken(); // 跳过 user 字段名 parser.nextToken(); // 跳过 user 对象开始 long userId parser.nextLongValue(0L); // user.id // ... 继续解析 } } }实测内存占用下降 65%GC 时间从 200ms 降到 30ms。4.5 故障熔断当上游挂了别拖垮整个 Flink job我们加了一个CircuitBreaker机制连续 5 次 HTTP 请求失败就标记该 subtask 为“熔断”停止拉取等待 5 分钟后自动恢复。实现很简单在run()方法里int consecutiveFailureCount 0; while (isRunning) { try { // 执行 HTTP 请求 consecutiveFailureCount 0; // 成功则清零 } catch (Exception e) { consecutiveFailureCount; if (consecutiveFailureCount 5) { LOG.error(Circuit breaker triggered for subtask {}, getRuntimeContext().getIndexOfThisSubtask()); Thread.sleep(5 * 60 * 1000); // 熔断 5 分钟 consecutiveFailureCount 0; } Thread.sleep(1000); } }这个机制让我们在某次上游服务雪崩时Flink job 依然保持 99.9% 的可用性只是该 subtask 的数据延迟了 5 分钟。4.6 日志分级让排错像读小说一样简单connector 的日志必须分三级INFO正常流程如 “Started reading from cursorabc123”WARN可恢复异常如 “HTTP 429, will retry in 1s”ERROR致命错误如 “Failed to deserialize JSON: invalid timestamp format”。我们禁用了 OkHttp 的 DEBUG 日志太 verbose只保留INFOLogger.getLogger(okhttp3).setLevel(Level.INFO);同时在HttpApiSource的run()方法里每个关键步骤都加日志LOG.info(Subtask {} starting HTTP request to {}, getRuntimeContext().getIndexOfThisSubtask(), url); LOG.debug(Request headers: {}, request.headers()); LOG.info(Subtask {} received {} records, getRuntimeContext().getIndexOfThisSubtask(), jsonArray.length());这样当问题发生时运维同学只需 grepSubtask 3就能串起完整链路不用翻几十个日志文件。5. 常见故障排查链路从报错日志定位到根因的完整路径在生产环境90% 的 connector 问题都集中在几个固定模式。我把它们整理成一条标准化排查链路每一步都附带真实日志样例和 root cause。5.1 现象SQL 作业启动失败日志报ClassNotFoundException: org.apache.flink.table.factories.DynamicTableFactory完整日志Caused by: java.lang.ClassNotFoundException: org.apache.flink.table.factories.DynamicTableFactory at java.net.URLClassLoader.findClass(URLClassLoader.java:387) at java.lang.ClassLoader.loadClass(ClassLoader.java:418) at org.apache.flink.util.FlinkUserCodeClassLoader.loadClass(FlinkUserCodeClassLoader.java:42)排查路径检查META-INF/services/org.apache.flink.table.factories.DynamicTableFactory文件是否存在如果存在用jar -tf your-connector.jar | grep META-INF确认它被打包进 JAR如果存在但还是报错检查pom.xml是否误把flink-table-api-javascope 设为compile应为provided最后检查 Flink 版本Flink 1.17 的 factory 接口和 1.18 不兼容必须严格匹配。根因SPI 文件未生成或依赖范围错误导致 Flink 无法发现你的 factory 类。5.2 现象作业运行中Web UI 显示Source并行度为 1但配置了parallelism.default8完整日志无明显错误但numRecordsInPerSecond指标始终为 0。排查路径登录 Web UI进入Job Overview→Vertex Metrics查看Source节点的numRecordsInPerSecond如果为 0检查HttpApiSource的run()方法是否进入了while循环在run()开头加LOG.info(Source subtask {} started, getRuntimeContext().getIndexOfThisSubtask())如果日志没输出说明SourceFunction没被实例化检查DynamicTableSource.getScanRuntimeProvider()是否返回了正确的SourceFunctionProvider。根因getScanRuntimeProvider()返回了null或SourceFunctionProvider.of()的第二个参数isParallel设为false。5.3 现象数据能读但WATERMARK不推进下游窗口一直不触发完整日志2024-04-05 10:23:45,123 INFO org.apache.flink.streaming.api.watermark.WatermarkGenerator - Current watermark: 1712345678000 2024-04-05 10:23:45,323