Any help is appreciated about the exception below.
Also my Kafkasource code is below. The parallelism is 16 for this task.
KafkaSource<String> sourceStationsPeriodic = KafkaSource.<
String>builder()
.setBootstrapServers(parameter.get(
KAFKA_SOURCE_STATIONS_BOOTLOADER_PROPERTY))
.setTopics(parameter.get(
KAFKA_TOPIC_READ_WIFI))
.setGroupId(parameter.get(KAFKA_GROUP))
.setStartingOffsets(OffsetsInitializer.
latest())
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build();
// Our Kafka Source
KafkaSource<String> sourceStationsWifiInterface =
KafkaSource.<String>builder()
.setBootstrapServers(parameter.get(
KAFKA_SOURCE_STATIONS_BOOTLOADER_PROPERTY))
.setTopics(parameter.get(
KAFKA_TOPIC_READ_WIFI_INTERFACE))
.setGroupId(parameter.get(KAFKA_GROUP))
.setStartingOffsets(OffsetsInitializer.
latest())
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build();
KafkaSource<String> sourceTwinMessage = KafkaSource.<String>
builder()
.setBootstrapServers(parameter.get(
KAFKA_SOURCE_STATIONS_BOOTLOADER_PROPERTY))
.setTopics(parameter.get(
KAFKA_TOPIC_READ_TWIN_MESSAGE))
.setGroupId(parameter.get(KAFKA_GROUP))
.setStartingOffsets(OffsetsInitializer.
latest())
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build();
KafkaSource<String> sourceStationsOnDemand = KafkaSource.<
String>builder()
.setBootstrapServers(parameter.get(
KAFKA_SOURCE_STATIONS_BOOTLOADER_PROPERTY))
.setTopics(parameter.get(
KAFKA_TOPIC_READ_STATIONS_ON_DEMAND))
.setGroupId(parameter.get(KAFKA_GROUP))
.setStartingOffsets(OffsetsInitializer.
latest())
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build();
KafkaSource<String> sourceDeviceInfo = KafkaSource.<String>
builder()
.setBootstrapServers(parameter.get(
KAFKA_SOURCE_STATIONS_BOOTLOADER_PROPERTY))
.setTopics(parameter.get(
KAFKA_TOPIC_READ_DEVICE_INFO))
.setGroupId(parameter.get(KAFKA_GROUP))
.setStartingOffsets(OffsetsInitializer.
latest())
.setValueOnlyDeserializer(new
SimpleStringSchema())
.build();
2023-07-23 07:06:24,927 WARN org.apache.kafka.common.utils.AppInfoParser
[] - Error registering AppInfo mbean
javax.management.InstanceAlreadyExistsException:
kafka.admin.client:type=app-info,id=wifialgogroup1-enumerator-admin-client
at com.sun.jmx.mbeanserver.Repository.addMBean(Repository.java:436)
~[?:?]
at
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerWithRepository(DefaultMBeanServerInterceptor.java:1855)
~[?:?]
at
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerDynamicMBean(DefaultMBeanServerInterceptor.java:955)
~[?:?]
at
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerObject(DefaultMBeanServerInterceptor.java:890)
~[?:?]
at
com.sun.jmx.interceptor.DefaultMBeanServerInterceptor.registerMBean(DefaultMBeanServerInterceptor.java:320)
~[?:?]
at
com.sun.jmx.mbeanserver.JmxMBeanServer.registerMBean(JmxMBeanServer.java:522)
~[?:?]
at
org.apache.kafka.common.utils.AppInfoParser.registerAppInfo(AppInfoParser.java:64)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.kafka.clients.admin.KafkaAdminClient.<init>(KafkaAdminClient.java:597)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:539)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.kafka.clients.admin.KafkaAdminClient.createInternal(KafkaAdminClient.java:478)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at org.apache.kafka.clients.admin.Admin.create(Admin.java:133)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.kafka.clients.admin.AdminClient.create(AdminClient.java:39)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.getKafkaAdminClient(KafkaSourceEnumerator.java:410)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.flink.connector.kafka.source.enumerator.KafkaSourceEnumerator.start(KafkaSourceEnumerator.java:151)
~[blob_p-7f823076a9b41619d082270330273927b5c89588-6161d6df252e036e35c93e64446c8834:?]
at
org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$start$1(SourceCoordinator.java:225)
~[flink-dist-1.16.0.jar:1.16.0]
at
org.apache.flink.runtime.source.coordinator.SourceCoordinator.lambda$runInEventLoop$10(SourceCoordinator.java:449)
~[flink-dist-1.16.0.jar:1.16.0]
at
org.apache.flink.util.ThrowableCatchingRunnable.run(ThrowableCatchingRunnable.java:40)
[flink-dist-1.16.0.jar:1.16.0]
at
java.util.concurrent.Executors$RunnableAdapter.call(Executors.java:515)
[?:?]
at java.util.concurrent.FutureTask.run(FutureTask.java:264) [?:?]
at
java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
[?:?]
at
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
[?:?]
at
java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
[?:?]
at java.lang.Thread.run(Thread.java:829) [?:?]
