qq77yes opened a new issue, #10870: URL: https://github.com/apache/seatunnel/issues/10870
### Search before asking - [x] I had searched in the [issues](https://github.com/apache/seatunnel/issues?q=is%3Aissue+label%3A%22bug%22) and found no similar issues. ### What happened 在v2.3.13中，通过seatunnel 往Milvus中导入数据一直报错。 数据embedding的执行结果是符合预期的，但是从debug日志中看，获取不到 context 无法初始化sink 进而初始化一个错误的milvus 表 ``` 2026-05-11 18:03:56,970 DEBUG [o.a.s.a.t.f.FactoryUtil ] [main] - java.lang.NullPointerException: Cannot invoke "org.apache.seatunnel.api.table.factory.TableSinkFactoryContext.getOptions()" because "context" is null at org.apache.seatunnel.connectors.seatunnel.milvus.sink.MilvusSinkFactory.createSink(MilvusSinkFactory.java:56) at org.apache.seatunnel.api.table.factory.FactoryUtil.lambda$createAndPrepareSink$1(FactoryUtil.java:213) at org.apache.seatunnel.api.table.factory.FactoryUtil.isFallback(FactoryUtil.java:458) at org.apache.seatunnel.api.table.factory.FactoryUtil.createAndPrepareSink(FactoryUtil.java:209) at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.createSinkAction(MultipleTableJobConfigParser.java:690) at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parseSink(MultipleTableJobConfigParser.java:613) at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parse(MultipleTableJobConfigParser.java:260) at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.getLogicalDag(ClientJobExecutionEnvironment.java:122) at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.execute(ClientJobExecutionEnvironment.java:190) at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:204) at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40) at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:40) 2026-05-11 18:03:56,977 INFO [o.a.s.a.t.f.FactoryUtil ] [main] - Create sink 'Milvus' with upstream input catalog-table[database: null, schema: null, table: raw_data] 2026-05-11 18:03:56,979 INFO [p.MultipleTableJobConfigParser] [main] - Unsupported multi table sink api, rollback to sink template ``` 执行中的上下文也发生错乱 `2026-05-11 18:03:57,526 INFO [a.s.a.s.SaveModeExecuteWrapper] [seatunnel-coordinator-service-2] - Executing save mode for table: data_goverment_test.raw_data, with SchemaSaveMode: ERROR_WHEN_SCHEMA_NOT_EXIST, DataSaveMode: APPEND_DATA using Catalog: S3File 2026-05-11 18:03:57,530 DEBUG [g.n.s.i.g.n.NettyClientHandler] [grpc-nio-worker-ELG-1-4] - [id: 0xe12e4b58, L:/172.20.3.24:51994 - R:/10.0.10.49:19530] OUTBOUND HEADERS: streamId=5 headers=GrpcHttp2OutboundHeaders[:authority: 10.0.10.49:19530, :path: /milvus.proto.milvus.MilvusService/HasCollection, :method: POST, :scheme: http, content-type: application/grpc, te: trailers, user-agent: grpc-java-netty/1.59.1, authorization: cm9vdDpNaWx2dXM=, dbname: default, grpc-accept-encoding: gzip] streamDependency=0 weight=16 exclusive=false padding=0 endStream=false` 进而报错 `The sink table not exist` ### SeaTunnel Version 2.3.13 ### SeaTunnel Config ```conf env { job.mode = "BATCH" parallelism = 1 job.name = "complex_json_from_minio_to_es" } # 第一步：从MinIO读取整个JSON文件作为单行文本数据 source { # 使用S3File连接器连接MinIO S3File { bucket = "s3a://qq-test" path = "/user_json/3ebe910703c14983bb44.json" access_key = "" secret_key = "x" fs.s3a.endpoint = "http://10.0.10.103:9000"; charset = "UTF-8" fs.s3a.aws.credentials.provider = "org.apache.hadoop.fs.s3a.SimpleAWSCredentialsProvider" file_format_type = "json" plugin_output = "raw_data" # 定义复杂的嵌套 Schema # SeaTunnel 会根据这个 Schema 解析 JSON 内容 schema = { fields { profile = string experience = string achievements = string relationships = string mediaAndNews = string uniq_id = string } } } } transform { Sql { plugin_input = ["raw_data"] query = """ select uniq_id, CAST(profile AS String) as PROFILE_STR from raw_data """ plugin_output = "stringified_data" } Embedding { plugin_input = ["stringified_data"] model_provider = OPENAI model = bge-m3 api_key = '' secret_key = '' api_path = "http://10.0.10.49:10003/v1/embeddings"; vectorization_fields { profile_vector = PROFILE_STR } plugin_output = "embedded_table" result_table_name = "embedded_table" } } sink { Milvus { url = "http://10.0.10.49:19530"; token = "root:Milvus" # 如果没设密码，默认通常是 root:Milvus 或为空 batch_size = 1000 plugin_input = "embedded_table" source_table_name = "embedded_table" database = "data_goverment_test" collection_name = "user_data" schema_save_mode = "ERROR_WHEN_SCHEMA_NOT_EXIST" collection_description = { uniq_id = "uniq_id" PROFILE_STR = "profile_data" # 建议在 Milvus 中设为 VARCHAR 或 JSON 类型 profile_vector = "profile_vector" # 必须yy是 FloatVector 类型 } } } ``` ### Running Command ```shell seatunnel --config xxx.conf -m local ``` ### Error Exception ```log 2026-05-11 18:03:56,962 INFO [.s.p.d.AbstractPluginDiscovery] [main] - find connector jar and dependency for PluginIdentifier{engineType='seatunnel', pluginType='sink', pluginName='Milvus'}: [file:/E:/software/apache-seatunnel-2.3.13-bin.tar/apache-seatunnel-2.3.13-bin/apache-seatunnel-2.3.13/connectors/connector-milvus-2.3.13.jar] 2026-05-11 18:03:56,970 DEBUG [o.a.s.a.t.f.FactoryUtil ] [main] - java.lang.NullPointerException: Cannot invoke "org.apache.seatunnel.api.table.factory.TableSinkFactoryContext.getOptions()" because "context" is null at org.apache.seatunnel.connectors.seatunnel.milvus.sink.MilvusSinkFactory.createSink(MilvusSinkFactory.java:56) at org.apache.seatunnel.api.table.factory.FactoryUtil.lambda$createAndPrepareSink$1(FactoryUtil.java:213) at org.apache.seatunnel.api.table.factory.FactoryUtil.isFallback(FactoryUtil.java:458) at org.apache.seatunnel.api.table.factory.FactoryUtil.createAndPrepareSink(FactoryUtil.java:209) at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.createSinkAction(MultipleTableJobConfigParser.java:690) at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parseSink(MultipleTableJobConfigParser.java:613) at org.apache.seatunnel.engine.core.parse.MultipleTableJobConfigParser.parse(MultipleTableJobConfigParser.java:260) at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.getLogicalDag(ClientJobExecutionEnvironment.java:122) at org.apache.seatunnel.engine.client.job.ClientJobExecutionEnvironment.execute(ClientJobExecutionEnvironment.java:190) at org.apache.seatunnel.core.starter.seatunnel.command.ClientExecuteCommand.execute(ClientExecuteCommand.java:204) at org.apache.seatunnel.core.starter.SeaTunnel.run(SeaTunnel.java:40) at org.apache.seatunnel.core.starter.seatunnel.SeaTunnelClient.main(SeaTunnelClient.java:40) 2026-05-11 18:03:56,977 INFO [o.a.s.a.t.f.FactoryUtil ] [main] - Create sink 'Milvus' with upstream input catalog-table[database: null, schema: null, table: raw_data] 2026-05-11 18:03:56,979 INFO [p.MultipleTableJobConfigParser] [main] - Unsupported multi table sink api, rollback to sink template ... 2026-05-11 18:03:57,516 DEBUG [g.n.s.i.g.n.NettyClientHandler] [grpc-nio-worker-ELG-1-4] - [id: 0xe12e4b58, L:/172.20.3.24:51994 - R:/10.0.10.49:19530] INBOUND DATA: streamId=3 padding=0 endStream=false length=113 bytes=000000006c0a00125e0a0776322e342e3131121c547565205365702031302030303a31333a33352055544320323032341a07633136363235332220676f207665... 2026-05-11 18:03:57,516 DEBUG [g.n.s.i.g.n.NettyClientHandler] [grpc-nio-worker-ELG-1-4] - [id: 0xe12e4b58, L:/172.20.3.24:51994 - R:/10.0.10.49:19530] OUTBOUND PING: ack=false bytes=1234 2026-05-11 18:03:57,518 DEBUG [g.n.s.i.g.n.NettyClientHandler] [grpc-nio-worker-ELG-1-4] - [id: 0xe12e4b58, L:/172.20.3.24:51994 - R:/10.0.10.49:19530] INBOUND HEADERS: streamId=3 headers=GrpcHttp2ResponseHeaders[grpc-status: 0, grpc-message: ] padding=0 endStream=true 2026-05-11 18:03:57,525 DEBUG [g.n.s.i.g.n.NettyClientHandler] [grpc-nio-worker-ELG-1-4] - [id: 0xe12e4b58, L:/172.20.3.24:51994 - R:/10.0.10.49:19530] INBOUND PING: ack=true bytes=1234 2026-05-11 18:03:57,526 INFO [a.s.a.s.SaveModeExecuteWrapper] [seatunnel-coordinator-service-2] - Executing save mode for table: data_goverment_test.raw_data, with SchemaSaveMode: ERROR_WHEN_SCHEMA_NOT_EXIST, DataSaveMode: APPEND_DATA using Catalog: S3File ... 2026-05-11 18:03:57,544 ERROR [o.a.s.e.s.CoordinatorService ] [seatunnel-coordinator-service-2] - [localhost]:5801 [seatunnel-405222] [5.1] submit job 1106159750325731329 error org.apache.seatunnel.common.exception.SeaTunnelRuntimeException: ErrorCode:[API-09], ErrorDescription:[Handle save mode failed] at org.apache.seatunnel.engine.server.master.JobMaster.handleSaveMode(JobMaster.java:576) at org.apache.seatunnel.engine.server.master.JobMaster.lambda$init$1(JobMaster.java:267) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.accept(ForEachOps.java:183) at java.base/java.util.stream.ReferencePipeline$2$1.accept(ReferencePipeline.java:177) at java.base/java.util.stream.ReferencePipeline$3$1.accept(ReferencePipeline.java:195) at java.base/java.util.Iterator.forEachRemaining(Iterator.java:133) at java.base/java.util.Spliterators$IteratorSpliterator.forEachRemaining(Spliterators.java:1801) at java.base/java.util.stream.AbstractPipeline.copyInto(AbstractPipeline.java:484) at java.base/java.util.stream.AbstractPipeline.wrapAndCopyInto(AbstractPipeline.java:474) at java.base/java.util.stream.ForEachOps$ForEachOp.evaluateSequential(ForEachOps.java:150) at java.base/java.util.stream.ForEachOps$ForEachOp$OfRef.evaluateSequential(ForEachOps.java:173) at java.base/java.util.stream.AbstractPipeline.evaluate(AbstractPipeline.java:234) at java.base/java.util.stream.ReferencePipeline.forEach(ReferencePipeline.java:497) at org.apache.seatunnel.engine.server.master.JobMaster.init(JobMaster.java:261) at org.apache.seatunnel.engine.server.CoordinatorService.lambda$submitJob$6(CoordinatorService.java:663) at org.apache.seatunnel.api.tracing.MDCRunnable.run(MDCRunnable.java:43) at java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1130) at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:630) at java.base/java.lang.Thread.run(Thread.java:832) Caused by: org.apache.seatunnel.common.exception.SeaTunnelRuntimeException: ErrorCode:[API-11], ErrorDescription:[The sink table not exist] - The sink table not exist at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.errorWhenSchemaNotExist(DefaultSaveModeHandler.java:145) at org.apache.seatunnel.api.sink.DefaultSaveModeHandler.handleSchemaSaveMode(DefaultSaveModeHandler.java:91) at org.apache.seatunnel.api.sink.SaveModeHandler.handleSaveMode(SaveModeHandler.java:42) at org.apache.seatunnel.api.sink.SaveModeExecuteWrapper.execute(SaveModeExecuteWrapper.java:36) at org.apache.seatunnel.engine.server.master.JobMaster.handleSaveMode(JobMaster.java:571) ... 20 more ``` ### Zeta or Flink or Spark Version zeta ### Java or Scala Version _No response_ ### Screenshots _No response_ ### Are you willing to submit PR? - [ ] Yes I am willing to submit a PR! ### Code of Conduct - [x] I agree to follow this project's [Code of Conduct](https://www.apache.org/foundation/policies/conduct) -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
