订阅kafka动态匹配表名问题

[ksdb1155228]

我创建的订阅任务，任务中指定单表insert \ update \delete 都没问题，但是不指定表名做动态匹配则无法消费kafka中消息。
请大佬们看看 是否我创建的任务语句有问题。
我创建的任务语句如下：
CREATE ROUTINE LOAD yunwei.test_kafka_032
WITH MERGE
DELETE ON crud = "d"
PROPERTIES
(
"format" = "json" ,
"desired_concurrent_number"="1",
"max_batch_interval" = "20",
"max_batch_rows" = "300000",
"max_batch_size" = "209715200",
"max_error_number" = "9999999999"
)
FROM KAFKA
(
"kafka_broker_list" = "10.4.33.234:9092",
"kafka_topic" = "testDorisKafkaTopic",
"property.group.id" = "test_kafka_group",
"property.client.id" = "test_kafka_client",
"property.kafka_default_offsets" = "OFFSET_BEGINNING"
)
发送的数据内容：
test_orders|{"id":0,"shop_code":"123456","shop_name":"测试店铺","order_id":666666666,"sku":"ABCD","quantity":8,"unit_price":88.88,"commit_time":"2023-12-29 09:42:39","crud":"r"}
查看任务的reason:
[INTERNAL_ERROR]empty data
0. /root/src/doris-2.0/be/src/common/stack_trace.cpp:302: StackTrace::tryCapture() @ 0x000000000ba1f197 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be

1. /root/src/doris-2.0/be/src/common/stack_trace.h:0: doris::get_stack_traceabi:cxx11 @ 0x000000000ba1d72d in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
2. /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/basic_string.h:187: doris::Status doris::Status::Error(int, std::basic_string_view<char, std::char_traits >) @ 0x000000000aeb6e2b in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
3. /root/src/doris-2.0/be/src/io/fs/multi_table_pipe.cpp:133: doris::io::MultiTablePipe::dispatch(std::__cxx11::basic_string<char, std::char_traits, std::allocator > const&, char const*, unsigned long, doris::Status (doris::io::KafkaConsumerPipe::)(char const, unsigned long)) @ 0x000000000b8c1fbb in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
4. /root/src/doris-2.0/be/src/io/fs/multi_table_pipe.cpp:0: doris::io::MultiTablePipe::append_json(char const*, unsigned long) @ 0x000000000b8c2726 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
5. /root/src/doris-2.0/be/src/common/status.h:432: doris::KafkaDataConsumerGroup::start_all(std::shared_ptrdoris::StreamLoadContext) @ 0x000000000b8bf6c2 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
6. /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/shared_ptr_base.h:701: doris::RoutineLoadTaskExecutor::exec_task(std::shared_ptrdoris::StreamLoadContext, doris::DataConsumerPool*, std::function<void (std::shared_ptrdoris::StreamLoadContext)>) @ 0x000000000b8b0d30 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
7. /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/std_function.h:244: std::_Function_handler<void (), std::_Bind_result<void, void (doris::RoutineLoadTaskExecutor::(doris::RoutineLoadTaskExecutor, std::shared_ptrdoris::StreamLoadContext, doris::DataConsumerPool*, doris::RoutineLoadTaskExecutor::submit_task(doris::TRoutineLoadTask const&)::$_0))(std::shared_ptrdoris::StreamLoadContext, doris::DataConsumerPool*, std::function<void (std::shared_ptrdoris::StreamLoadContext)>)> >::_M_invoke(std::_Any_data const&) @ 0x000000000b8b26c9 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
8. /var/local/ldb-toolchain/bin/../lib/gcc/x86_64-linux-gnu/11/../../../../include/c++/11/bits/atomic_base.h:646: doris::WorkThreadPool::work_thread(int) @ 0x000000000aec0a32 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
9. /data/gcc-11.1.0/build/x86_64-pc-linux-gnu/libstdc++-v3/include/bits/unique_ptr.h:85: execute_native_thread_routine @ 0x00000000155b2fb0 in /opt/apache-doris-2.0.2-bin-x64/be/lib/doris_be
10. start_thread @ 0x0000000000007dd5 in /usr/lib64/libpthread-2.17.so
11. clone @ 0x00000000000fdead in /usr/lib64/libc-2.17.so

[sollhui]

Is table exist in Doris? Using single stream to load multi table should promise all table exist.For example, load a data like: test|a,b,c. If test do not exist, load job will do no consume.