[ https://issues.apache.org/jira/browse/FLINK-32246?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=17728937#comment-17728937 ]
jeff-zou edited comment on FLINK-32246 at 6/3/23 11:19 AM: ----------------------------------------------------------- [~aitozi] No, It's run in the Native Kubernates. The code like below: {code:java} StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); tEnv.executeSql("create table kafka_source() with ('connector'='kafka')"); tEnv.executeSql("insert into sink_table1 select * from kafka_source;"); TableResult tableResult= tEnv.executeSql("insert into sink_table2 select * from kafka_source;"); tableResult.getJobClient().get().getJobExecutionResult().get(); {code} was (Author: JIRAUSER282259): No, It's run in the Native Kubernates. The code like below: {code:java} StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment(); StreamTableEnvironment tEnv = StreamTableEnvironment.create(env); tEnv.executeSql("create table kafka_source() with ('connector'='kafka')"); tEnv.executeSql("insert into sink_table1 select * from kafka_source;"); TableResult tableResult= tEnv.executeSql("insert into sink_table2 select * from kafka_source;"); tableResult.getJobClient().get().getJobExecutionResult().get(); {code} > javax.management.InstanceAlreadyExistsException > ----------------------------------------------- > > Key: FLINK-32246 > URL: https://issues.apache.org/jira/browse/FLINK-32246 > Project: Flink > Issue Type: Bug > Components: Connectors / Kafka > Affects Versions: 1.15.2 > Reporter: jeff-zou > Priority: Major > > Flink SQL throws an > exception(javax.management.InstanceAlreadyExistsException) when trying to > perform multiple sink operations on the same kafka source . > > sql example: > {code:java} > create table kafka_source() with ('connector'='kafka'); > insert into sink_table1 select * from kafka_source; > insert into sink_table2 select * from kafka_source; {code} > The Exception as below: > {code:java} > javax.management.InstanceAlreadyExistsException: > kafka.admin.client:type=app-info,id=***** > > java.management/com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:436) > > java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1855) > > java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:955) > > java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:890) > > java.management/com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:320) > > java.management/com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522) > > org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:64) > > org.apache.kafka.clients.admin.KafkaAdminClient.<init>(KafkaAdminClient.java:500) > > org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:444) > org.apache.kafka.clients.admin.Admin.create(Admin.java:59) > org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:39) > > org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.getKafkaAdminClient(KafkaSourceEnumerator.java:410) > > org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.start(KafkaSourceEnumerator.java:151) > > org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$start$1(SourceCoordinator.java:209) > > org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$9(SourceCoordinator.java:406) > > org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40) > > java.base/java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515) > java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264) > > java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304) > > java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128) > > java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628) > java.base/java.lang.Thread.run(Thread.java:829) {code} -- This message was sent by Atlassian Jira (v8.20.10#820010)