sql示例: create table kafka_source() with ('connector'='kafka'); insert into sink_table1 select * from kafka_source; insert into sink_table2 select * from kafka_source;
报错内容如下: javax.management.InstanceAlreadyExistsException: kafka.admin.client:type=app-info,id=jxqy_customer_service-enumerator-admin-client \tat java.management/com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:436) \tat java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1855) \tat java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:955) \tat java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:890) \tat java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:320) \tat java.management/com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) \tat org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:64) \tat org.apache.kafka.clients.admin.KafkaAdminClient.<init>(KafkaAdminClient.java:500) \tat org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:444) \tat org.apache.kafka.clients.admin.Admin.create(Admin.java:59) \tat org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:39) \tat org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.getKafkaAdminClient(KafkaSourceEnumerator.java:410) \tat org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.start(KafkaSourceEnumerator.java:151) \tat org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$start$1(SourceCoordinator.java:209) \tat org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$9(SourceCoordinator.java:406) \tat org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) \tat java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) \tat java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) \tat java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) \tat java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) \tat java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) \tat java.base/java.lang.Thread.run(Thread.java:829)