azure-sdk-for-java: [BUG]AmqpException: New receiver 'P1-b177db87-de2c-481f-8107-f9262202a6a6' with higher epoch of '0' is created hence current receiver
Describe the bug
My application is listening to a EventHub for messages. When I message is received, I ingest that message into ADX. If there are a error or unknown message type, I’ll produce a message in another Eventhub.
When I launch the application, the messages are received and saved into ADX without issue, but something I get those errors and the application stop receiving incomming messages.
I’m using a ManagedStreamingIngestClient to ingest data into ADX.
Here the logs (I removed all the messages sent… around 50+)
C:\dev\jdk11\bin\java.exe -XX:TieredStopAtLevel=1 -noverify -Dspring.output.ansi.enabled=always -Dcom.sun.management.jmxremote -Dspring.jmx.enabled=true -Dspring.liveBeansView.mbeanDomain -Dspring.application.admin.enabled=true "-javaagent:C:\Program Files\JetBrains\IntelliJ IDEA 221.5080.93\lib\idea_rt.jar=56659:C:\Program Files\JetBrains\IntelliJ IDEA 221.5080.93\bin" -Dfile.encoding=UTF-8 -classpath C:\workspace\azure\poc-azure\iot-to-adx\target\classes;C:\Users\sd003526\.m2\repository\com\microsoft\azure\sdk\iot\iot-device-client\2.1.2\iot-device-client-2.1.2.jar;C:\Users\sd003526\.m2\repository\org\apache\qpid\proton-j\0.34.0\proton-j-0.34.0.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\qpid-proton-j-extensions\1.2.4\qpid-proton-j-extensions-1.2.4.jar;C:\Users\sd003526\.m2\repository\org\eclipse\paho\org.eclipse.paho.client.mqttv3\1.2.5\org.eclipse.paho.client.mqttv3-1.2.5.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\sdk\iot\provisioning\security\security-provider\2.0.0\security-provider-2.0.0.jar;C:\Users\sd003526\.m2\repository\com\google\code\gson\gson\2.9.1\gson-2.9.1.jar;C:\Users\sd003526\.m2\repository\org\slf4j\slf4j-api\1.7.36\slf4j-api-1.7.36.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\iothub-java-client\iothub-java-service-client\1.0.11\iothub-java-service-client-1.0.11.jar;C:\Users\sd003526\.m2\repository\org\glassfish\javax.json\1.0.4\javax.json-1.0.4.jar;C:\Users\sd003526\.m2\repository\commons-codec\commons-codec\1.15\commons-codec-1.15.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\iothub-java-client\websocket-transport-layer\0.1.1\websocket-transport-layer-0.1.1.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\kusto\kusto-data\3.2.0\kusto-data-3.2.0.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\msal4j\1.11.0\msal4j-1.11.0.jar;C:\Users\sd003526\.m2\repository\com\nimbusds\oauth2-oidc-sdk\9.7\oauth2-oidc-sdk-9.7.jar;C:\Users\sd003526\.m2\repository\com\github\stephenc\jcip\jcip-annotations\1.0-1\jcip-annotations-1.0-1.jar;C:\Users\sd003526\.m2\repository\com\nimbusds\content-type\2.1\content-type-2.1.jar;C:\Users\sd003526\.m2\repository\com\nimbusds\lang-tag\1.5\lang-tag-1.5.jar;C:\Users\sd003526\.m2\repository\com\nimbusds\nimbus-jose-jwt\9.9.3\nimbus-jose-jwt-9.9.3.jar;C:\Users\sd003526\.m2\repository\org\apache\httpcomponents\httpclient\4.5.13\httpclient-4.5.13.jar;C:\Users\sd003526\.m2\repository\org\apache\httpcomponents\httpcore\4.4.15\httpcore-4.4.15.jar;C:\Users\sd003526\.m2\repository\org\json\json\20201115\json-20201115.jar;C:\Users\sd003526\.m2\repository\org\jetbrains\annotations\22.0.0\annotations-22.0.0.jar;C:\Users\sd003526\.m2\repository\org\apache\commons\commons-text\1.9\commons-text-1.9.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-identity\1.6.0\azure-identity-1.6.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-core-http-netty\1.12.5\azure-core-http-netty-1.12.5.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-handler\4.1.79.Final\netty-handler-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-common\4.1.79.Final\netty-common-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-resolver\4.1.79.Final\netty-resolver-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-transport\4.1.79.Final\netty-transport-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-codec\4.1.79.Final\netty-codec-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-handler-proxy\4.1.79.Final\netty-handler-proxy-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-codec-socks\4.1.79.Final\netty-codec-socks-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-buffer\4.1.79.Final\netty-buffer-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-codec-http\4.1.79.Final\netty-codec-http-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-codec-http2\4.1.79.Final\netty-codec-http2-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-transport-native-unix-common\4.1.79.Final\netty-transport-native-unix-common-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-transport-native-epoll\4.1.79.Final\netty-transport-native-epoll-4.1.79.Final-linux-x86_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-transport-classes-epoll\4.1.79.Final\netty-transport-classes-epoll-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-transport-native-kqueue\4.1.79.Final\netty-transport-native-kqueue-4.1.79.Final-osx-x86_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-transport-classes-kqueue\4.1.79.Final\netty-transport-classes-kqueue-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-boringssl-static\2.0.53.Final\netty-tcnative-boringssl-static-2.0.53.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-classes\2.0.53.Final\netty-tcnative-classes-2.0.53.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-boringssl-static\2.0.53.Final\netty-tcnative-boringssl-static-2.0.53.Final-linux-x86_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-boringssl-static\2.0.53.Final\netty-tcnative-boringssl-static-2.0.53.Final-linux-aarch_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-boringssl-static\2.0.53.Final\netty-tcnative-boringssl-static-2.0.53.Final-osx-x86_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-boringssl-static\2.0.53.Final\netty-tcnative-boringssl-static-2.0.53.Final-osx-aarch_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-tcnative-boringssl-static\2.0.53.Final\netty-tcnative-boringssl-static-2.0.53.Final-windows-x86_64.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\msal4j-persistence-extension\1.1.0\msal4j-persistence-extension-1.1.0.jar;C:\Users\sd003526\.m2\repository\net\java\dev\jna\jna\5.5.0\jna-5.5.0.jar;C:\Users\sd003526\.m2\repository\net\java\dev\jna\jna-platform\5.6.0\jna-platform-5.6.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-core\1.32.0\azure-core-1.32.0.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\core\jackson-core\2.13.3\jackson-core-2.13.3.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\datatype\jackson-datatype-jsr310\2.13.3\jackson-datatype-jsr310-2.13.3.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\dataformat\jackson-dataformat-xml\2.13.3\jackson-dataformat-xml-2.13.3.jar;C:\Users\sd003526\.m2\repository\org\codehaus\woodstox\stax2-api\4.2.1\stax2-api-4.2.1.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\woodstox\woodstox-core\6.2.7\woodstox-core-6.2.7.jar;C:\Users\sd003526\.m2\repository\io\projectreactor\reactor-core\3.4.22\reactor-core-3.4.22.jar;C:\Users\sd003526\.m2\repository\org\reactivestreams\reactive-streams\1.0.4\reactive-streams-1.0.4.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\core\jackson-databind\2.13.3\jackson-databind-2.13.3.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\core\jackson-annotations\2.13.3\jackson-annotations-2.13.3.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\kusto\kusto-ingest\3.2.0\kusto-ingest-3.2.0.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\azure-storage\8.6.6\azure-storage-8.6.6.jar;C:\Users\sd003526\.m2\repository\com\microsoft\azure\azure-keyvault-core\1.2.4\azure-keyvault-core-1.2.4.jar;C:\Users\sd003526\.m2\repository\com\google\guava\guava\24.1.1-jre\guava-24.1.1-jre.jar;C:\Users\sd003526\.m2\repository\com\google\code\findbugs\jsr305\1.3.9\jsr305-1.3.9.jar;C:\Users\sd003526\.m2\repository\org\checkerframework\checker-compat-qual\2.0.0\checker-compat-qual-2.0.0.jar;C:\Users\sd003526\.m2\repository\com\google\errorprone\error_prone_annotations\2.1.3\error_prone_annotations-2.1.3.jar;C:\Users\sd003526\.m2\repository\com\google\j2objc\j2objc-annotations\1.1\j2objc-annotations-1.1.jar;C:\Users\sd003526\.m2\repository\org\codehaus\mojo\animal-sniffer-annotations\1.14\animal-sniffer-annotations-1.14.jar;C:\Users\sd003526\.m2\repository\com\univocity\univocity-parsers\2.9.1\univocity-parsers-2.9.1.jar;C:\Users\sd003526\.m2\repository\io\github\resilience4j\resilience4j-retry\1.7.1\resilience4j-retry-1.7.1.jar;C:\Users\sd003526\.m2\repository\io\github\resilience4j\resilience4j-core\1.7.1\resilience4j-core-1.7.1.jar;C:\Users\sd003526\.m2\repository\io\vavr\vavr\0.10.2\vavr-0.10.2.jar;C:\Users\sd003526\.m2\repository\io\vavr\vavr-match\0.10.2\vavr-match-0.10.2.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter-actuator\2.7.3\spring-boot-starter-actuator-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter\2.7.3\spring-boot-starter-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot\2.7.3\spring-boot-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-context\5.3.22\spring-context-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-aop\5.3.22\spring-aop-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-expression\5.3.22\spring-expression-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-autoconfigure\2.7.3\spring-boot-autoconfigure-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter-logging\2.7.3\spring-boot-starter-logging-2.7.3.jar;C:\Users\sd003526\.m2\repository\ch\qos\logback\logback-classic\1.2.11\logback-classic-1.2.11.jar;C:\Users\sd003526\.m2\repository\ch\qos\logback\logback-core\1.2.11\logback-core-1.2.11.jar;C:\Users\sd003526\.m2\repository\org\apache\logging\log4j\log4j-to-slf4j\2.17.2\log4j-to-slf4j-2.17.2.jar;C:\Users\sd003526\.m2\repository\org\apache\logging\log4j\log4j-api\2.17.2\log4j-api-2.17.2.jar;C:\Users\sd003526\.m2\repository\org\slf4j\jul-to-slf4j\1.7.36\jul-to-slf4j-1.7.36.jar;C:\Users\sd003526\.m2\repository\jakarta\annotation\jakarta.annotation-api\1.3.5\jakarta.annotation-api-1.3.5.jar;C:\Users\sd003526\.m2\repository\org\yaml\snakeyaml\1.30\snakeyaml-1.30.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-actuator-autoconfigure\2.7.3\spring-boot-actuator-autoconfigure-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-actuator\2.7.3\spring-boot-actuator-2.7.3.jar;C:\Users\sd003526\.m2\repository\io\micrometer\micrometer-core\1.9.3\micrometer-core-1.9.3.jar;C:\Users\sd003526\.m2\repository\org\hdrhistogram\HdrHistogram\2.1.12\HdrHistogram-2.1.12.jar;C:\Users\sd003526\.m2\repository\org\latencyutils\LatencyUtils\2.0.3\LatencyUtils-2.0.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter-webflux\2.7.3\spring-boot-starter-webflux-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter-json\2.7.3\spring-boot-starter-json-2.7.3.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\datatype\jackson-datatype-jdk8\2.13.3\jackson-datatype-jdk8-2.13.3.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\module\jackson-module-parameter-names\2.13.3\jackson-module-parameter-names-2.13.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter-reactor-netty\2.7.3\spring-boot-starter-reactor-netty-2.7.3.jar;C:\Users\sd003526\.m2\repository\io\projectreactor\netty\reactor-netty-http\1.0.22\reactor-netty-http-1.0.22.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-resolver-dns\4.1.79.Final\netty-resolver-dns-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-codec-dns\4.1.79.Final\netty-codec-dns-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-resolver-dns-native-macos\4.1.79.Final\netty-resolver-dns-native-macos-4.1.79.Final-osx-x86_64.jar;C:\Users\sd003526\.m2\repository\io\netty\netty-resolver-dns-classes-macos\4.1.79.Final\netty-resolver-dns-classes-macos-4.1.79.Final.jar;C:\Users\sd003526\.m2\repository\io\projectreactor\netty\reactor-netty-core\1.0.22\reactor-netty-core-1.0.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-web\5.3.22\spring-web-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-beans\5.3.22\spring-beans-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-webflux\5.3.22\spring-webflux-5.3.22.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-starter\4.4.0\spring-cloud-azure-starter-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-autoconfigure\4.4.0\spring-cloud-azure-autoconfigure-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-service\4.4.0\spring-cloud-azure-service-4.4.0.jar;C:\Users\sd003526\.m2\repository\jakarta\validation\jakarta.validation-api\2.0.2\jakarta.validation-api-2.0.2.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-starter-actuator\4.4.0\spring-cloud-azure-starter-actuator-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-actuator-autoconfigure\4.4.0\spring-cloud-azure-actuator-autoconfigure-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-actuator\4.4.0\spring-cloud-azure-actuator-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-stream-binder-eventhubs\4.4.0\spring-cloud-azure-stream-binder-eventhubs-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-stream-binder-eventhubs-core\4.4.0\spring-cloud-azure-stream-binder-eventhubs-core-4.4.0.jar;C:\Users\sd003526\.m2\repository\org\springframework\cloud\spring-cloud-stream\3.2.4\spring-cloud-stream-3.2.4.jar;C:\Users\sd003526\.m2\repository\org\springframework\boot\spring-boot-starter-validation\2.7.3\spring-boot-starter-validation-2.7.3.jar;C:\Users\sd003526\.m2\repository\org\apache\tomcat\embed\tomcat-embed-el\9.0.65\tomcat-embed-el-9.0.65.jar;C:\Users\sd003526\.m2\repository\org\hibernate\validator\hibernate-validator\6.2.4.Final\hibernate-validator-6.2.4.Final.jar;C:\Users\sd003526\.m2\repository\org\jboss\logging\jboss-logging\3.4.3.Final\jboss-logging-3.4.3.Final.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\classmate\1.5.1\classmate-1.5.1.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-messaging\5.3.22\spring-messaging-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\integration\spring-integration-jmx\5.5.14\spring-integration-jmx-5.5.14.jar;C:\Users\sd003526\.m2\repository\org\springframework\retry\spring-retry\1.3.3\spring-retry-1.3.3.jar;C:\Users\sd003526\.m2\repository\org\springframework\cloud\spring-cloud-function-context\3.2.5\spring-cloud-function-context-3.2.5.jar;C:\Users\sd003526\.m2\repository\net\jodah\typetools\0.6.2\typetools-0.6.2.jar;C:\Users\sd003526\.m2\repository\org\springframework\cloud\spring-cloud-function-core\3.2.5\spring-cloud-function-core-3.2.5.jar;C:\Users\sd003526\.m2\repository\javax\annotation\javax.annotation-api\1.3.2\javax.annotation-api-1.3.2.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-integration-azure-eventhubs\4.4.0\spring-integration-azure-eventhubs-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-integration-azure-core\4.4.0\spring-integration-azure-core-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-messaging-azure\4.4.0\spring-messaging-azure-4.4.0.jar;C:\Users\sd003526\.m2\repository\org\springframework\integration\spring-integration-core\5.5.14\spring-integration-core-5.5.14.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-tx\5.3.22\spring-tx-5.3.22.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-messaging-azure-eventhubs\4.4.0\spring-messaging-azure-eventhubs-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-messaging-eventhubs-checkpointstore-blob\1.15.1\azure-messaging-eventhubs-checkpointstore-blob-1.15.1.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-messaging-eventhubs\5.13.1\azure-messaging-eventhubs-5.13.1.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-core-amqp\2.7.1\azure-core-amqp-2.7.1.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-storage-blob\12.19.1\azure-storage-blob-12.19.1.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-storage-common\12.18.1\azure-storage-common-12.18.1.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-storage-internal-avro\12.4.2\azure-storage-internal-avro-12.4.2.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-resourcemanager\4.4.0\spring-cloud-azure-resourcemanager-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\spring\spring-cloud-azure-core\4.4.0\spring-cloud-azure-core-4.4.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-core-management\1.8.0\azure-core-management-1.8.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager\2.19.0\azure-resourcemanager-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-resources\2.19.0\azure-resourcemanager-resources-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-storage\2.19.0\azure-resourcemanager-storage-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-compute\2.19.0\azure-resourcemanager-compute-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-network\2.19.0\azure-resourcemanager-network-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-keyvault\2.19.0\azure-resourcemanager-keyvault-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-security-keyvault-keys\4.5.0\azure-security-keyvault-keys-4.5.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-security-keyvault-secrets\4.5.0\azure-security-keyvault-secrets-4.5.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-msi\2.19.0\azure-resourcemanager-msi-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-sql\2.19.0\azure-resourcemanager-sql-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-authorization\2.19.0\azure-resourcemanager-authorization-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-appservice\2.19.0\azure-resourcemanager-appservice-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-cosmos\2.19.0\azure-resourcemanager-cosmos-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-containerservice\2.19.0\azure-resourcemanager-containerservice-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-monitor\2.19.0\azure-resourcemanager-monitor-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-containerregistry\2.19.0\azure-resourcemanager-containerregistry-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-dns\2.19.0\azure-resourcemanager-dns-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-appplatform\2.19.0\azure-resourcemanager-appplatform-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\azure-storage-file-share\12.15.1\azure-storage-file-share-12.15.1.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-containerinstance\2.19.0\azure-resourcemanager-containerinstance-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-privatedns\2.19.0\azure-resourcemanager-privatedns-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-redis\2.19.0\azure-resourcemanager-redis-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-eventhubs\2.19.0\azure-resourcemanager-eventhubs-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-trafficmanager\2.19.0\azure-resourcemanager-trafficmanager-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-servicebus\2.19.0\azure-resourcemanager-servicebus-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-cdn\2.19.0\azure-resourcemanager-cdn-2.19.0.jar;C:\Users\sd003526\.m2\repository\com\azure\resourcemanager\azure-resourcemanager-search\2.19.0\azure-resourcemanager-search-2.19.0.jar;C:\Users\sd003526\.m2\repository\org\projectlombok\lombok\1.18.24\lombok-1.18.24.jar;C:\Users\sd003526\.m2\repository\org\apache\commons\commons-lang3\3.12.0\commons-lang3-3.12.0.jar;C:\Users\sd003526\.m2\repository\commons-beanutils\commons-beanutils\1.9.4\commons-beanutils-1.9.4.jar;C:\Users\sd003526\.m2\repository\commons-logging\commons-logging\1.2\commons-logging-1.2.jar;C:\Users\sd003526\.m2\repository\commons-collections\commons-collections\3.2.2\commons-collections-3.2.2.jar;C:\Users\sd003526\.m2\repository\org\springdoc\springdoc-openapi-webflux-ui\1.6.12\springdoc-openapi-webflux-ui-1.6.12.jar;C:\Users\sd003526\.m2\repository\org\springdoc\springdoc-openapi-webflux-core\1.6.12\springdoc-openapi-webflux-core-1.6.12.jar;C:\Users\sd003526\.m2\repository\org\springdoc\springdoc-openapi-common\1.6.12\springdoc-openapi-common-1.6.12.jar;C:\Users\sd003526\.m2\repository\io\swagger\core\v3\swagger-core\2.2.4\swagger-core-2.2.4.jar;C:\Users\sd003526\.m2\repository\com\fasterxml\jackson\dataformat\jackson-dataformat-yaml\2.13.3\jackson-dataformat-yaml-2.13.3.jar;C:\Users\sd003526\.m2\repository\io\swagger\core\v3\swagger-annotations\2.2.4\swagger-annotations-2.2.4.jar;C:\Users\sd003526\.m2\repository\io\swagger\core\v3\swagger-models\2.2.4\swagger-models-2.2.4.jar;C:\Users\sd003526\.m2\repository\org\webjars\swagger-ui\4.14.3\swagger-ui-4.14.3.jar;C:\Users\sd003526\.m2\repository\org\webjars\webjars-locator-core\0.50\webjars-locator-core-0.50.jar;C:\Users\sd003526\.m2\repository\io\github\classgraph\classgraph\4.8.149\classgraph-4.8.149.jar;C:\Users\sd003526\.m2\repository\net\minidev\json-smart\2.4.8\json-smart-2.4.8.jar;C:\Users\sd003526\.m2\repository\net\minidev\accessors-smart\2.4.8\accessors-smart-2.4.8.jar;C:\Users\sd003526\.m2\repository\org\ow2\asm\asm\9.1\asm-9.1.jar;C:\Users\sd003526\.m2\repository\jakarta\xml\bind\jakarta.xml.bind-api\2.3.3\jakarta.xml.bind-api-2.3.3.jar;C:\Users\sd003526\.m2\repository\jakarta\activation\jakarta.activation-api\1.2.2\jakarta.activation-api-1.2.2.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-core\5.3.22\spring-core-5.3.22.jar;C:\Users\sd003526\.m2\repository\org\springframework\spring-jcl\5.3.22\spring-jcl-5.3.22.jar com.comact.iothub.Application
. ____ _ __ _ _
/\\ / ___'_ __ _ _(_)_ __ __ _ \ \ \ \
( ( )\___ | '_ | '_| | '_ \/ _` | \ \ \ \
\\/ ___)| |_)| | | | | || (_| | ) ) ) )
' |____| .__|_| |_|_| |_\__, | / / / /
=========|_|==============|___/=/_/_/_/
:: Spring Boot :: (v2.7.3)
13:06:05.934 [background-preinit] INFO o.h.validator.internal.util.Version - HV000001: Hibernate Validator 6.2.4.Final
13:06:05.934 [main] INFO com.comact.iothub.Application - Starting Application using Java 11.0.15 on BID01795 with PID 13424 (C:\workspace\azure\poc-azure\iot-to-adx\target\classes started by sd003526 in C:\workspace\azure\poc-azure\azure-templates)
13:06:05.938 [main] INFO com.comact.iothub.Application - No active profile set, falling back to 1 default profile: "default"
13:06:06.635 [main] INFO o.s.i.c.DefaultConfiguringBeanFactoryPostProcessor - No bean named 'errorChannel' has been explicitly defined. Therefore, a default PublishSubscribeChannel will be created.
13:06:06.643 [main] INFO o.s.i.c.DefaultConfiguringBeanFactoryPostProcessor - No bean named 'integrationHeaderChannelRegistry' has been explicitly defined. Therefore, a default DefaultHeaderChannelRegistry will be created.
13:06:06.731 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'org.springframework.integration.config.IntegrationManagementConfiguration' of type [org.springframework.integration.config.IntegrationManagementConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.732 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration$IntegrationJmxConfiguration' of type [org.springframework.boot.autoconfigure.integration.IntegrationAutoConfiguration$IntegrationJmxConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.736 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'org.springframework.cloud.stream.config.BindersHealthIndicatorAutoConfiguration' of type [org.springframework.cloud.stream.config.BindersHealthIndicatorAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.737 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'bindersHealthContributor' of type [org.springframework.cloud.stream.config.BindersHealthIndicatorAutoConfiguration$BindersHealthContributor] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.738 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'bindersHealthIndicatorListener' of type [org.springframework.cloud.stream.config.BindersHealthIndicatorAutoConfiguration$BindersHealthIndicatorListener] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.739 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'BindingHandlerAdvise' of type [org.springframework.cloud.stream.config.BindingHandlerAdvise] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.742 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'spelConverter' of type [org.springframework.cloud.stream.config.SpelExpressionConverterConfiguration$SpelConverter] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.744 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'spring.jmx-org.springframework.boot.autoconfigure.jmx.JmxProperties' of type [org.springframework.boot.autoconfigure.jmx.JmxProperties] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.748 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration' of type [org.springframework.boot.autoconfigure.jmx.JmxAutoConfiguration] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.750 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'mbeanServer' of type [com.sun.jmx.mbeanserver.JmxMBeanServer] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.756 [main] INFO o.s.c.s.PostProcessorRegistrationDelegate$BeanPostProcessorChecker - Bean 'integrationChannelResolver' of type [org.springframework.integration.support.channel.BeanFactoryChannelResolver] is not eligible for getting processed by all BeanPostProcessors (for example: not eligible for auto-proxying)
13:06:06.895 [main] INFO c.m.a.k.i.ManagedStreamingIngestClient - Creating a new ManagedStreamingIngestClient from connection strings
13:06:06.896 [main] INFO c.m.a.k.i.QueuedIngestClientImpl - Creating a new IngestClient
13:06:06.997 [Timer-0] INFO c.m.a.kusto.ingest.ResourceManager - Refreshing Ingestion Auth Token
13:06:06.998 [main] INFO c.m.a.k.ingest.StreamingIngestClient - Creating a new StreamingIngestClient
13:06:07.276 [main] INFO o.s.b.a.e.web.EndpointLinksResolver - Exposing 17 endpoint(s) beneath base path '/actuator'
13:06:07.719 [Timer-0] INFO c.m.a.kusto.ingest.ResourceManager - Refreshing Ingestion Resources
13:06:07.740 [Timer-0] INFO c.m.a.kusto.ingest.ResourceManager - Refreshing Ingestion Resources Finised
13:06:08.277 [main] INFO o.s.c.s.m.DirectWithAttributesChannel - Channel 'application.devicemessages-in-0' has 1 subscriber(s).
13:06:08.435 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageChannel produce-out-0
13:06:08.440 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageChannel errorChannel
13:06:08.443 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageChannel devicemessages-in-0
13:06:08.445 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageChannel nullChannel
13:06:08.448 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageChannel produce_integrationflow.channel#0
13:06:08.452 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageHandler produce_integrationflow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0
13:06:08.464 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageHandler _org.springframework.integration.errorLogger
13:06:08.469 [main] INFO o.s.i.endpoint.EventDrivenConsumer - Adding {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
13:06:08.469 [main] INFO o.s.i.c.PublishSubscribeChannel - Channel 'application.errorChannel' has 1 subscriber(s).
13:06:08.469 [main] INFO o.s.i.endpoint.EventDrivenConsumer - started bean '_org.springframework.integration.errorLogger'
13:06:08.483 [main] INFO o.s.i.e.ReactiveStreamsConsumer - started bean 'produce_integrationflow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'
13:06:08.483 [main] INFO o.s.c.s.binder.DefaultBinderFactory - Creating binder: eventhubs
13:06:08.608 [main] INFO o.s.c.s.binder.DefaultBinderFactory - Caching the binder: eventhubs
13:06:08.609 [main] INFO o.s.c.s.binder.DefaultBinderFactory - Retrieving cached binder: eventhubs
13:06:08.674 [main] INFO o.s.c.s.m.DirectWithAttributesChannel - Channel 'application.produce-out-0' has 1 subscriber(s).
13:06:08.675 [main] INFO o.s.c.s.binder.DefaultBinderFactory - Retrieving cached binder: eventhubs
13:06:08.718 [main] INFO o.s.c.s.binder.BinderErrorChannel - Channel 'event-hub-adx.$Default.errors' has 1 subscriber(s).
13:06:08.719 [main] INFO o.s.c.s.binder.BinderErrorChannel - Channel 'event-hub-adx.$Default.errors' has 0 subscriber(s).
13:06:08.719 [main] INFO o.s.c.s.binder.BinderErrorChannel - Channel 'event-hub-adx.$Default.errors' has 1 subscriber(s).
13:06:08.719 [main] INFO o.s.c.s.binder.BinderErrorChannel - Channel 'event-hub-adx.$Default.errors' has 2 subscriber(s).
13:06:09.053 [main] INFO o.s.b.w.e.netty.NettyWebServer - Netty started on port 8080
13:06:09.062 [main] INFO o.s.i.m.IntegrationMBeanExporter - Registering MessageChannel event-hub-adx.$Default.errors
13:06:09.075 [main] INFO com.comact.iothub.Application - Started Application in 3.499 seconds (JVM running for 4.304)
13:06:11.434 [partition-pump-1-3] INFO c.c.i.c.TelemetryMessageConsumer - Consumed a new message [{"machine":"TrimExpert","scanner":"Trimmer","scannerId":"1","kpiPattern":"board-width-offset-mean-per-width","kpiDefinition":"4\u0026quot;","context":"Shift","contextId":"6","label":"Day Shift Nov 9, 2022","value":0.034595793701120287,"target":0.0,"min":-2.0,"max":2.0,"timestamp":"2022-11-09T16:38:00+0000","tstamp":1668011880000,"persistent":false,"objective":"{\"key\":\"default\",\"title\":\"default\",\"min\":-2.0,\"max\":2.0,\"target\":0.0,\"zones\":[],\"batchId\":null}","order":3.9139224265482824,"telemetryType":"KPI","customer":"Biewer","site":"Winona","plant":"Sawmill","subject":"Shift::Trimmer::board-width-offset-mean-per-width::4\u0026quot;"}] with headers [{azure_eventhubs_sequence_number=107824543, iothub-creation-time-utc=2022-11-09T16:38:27.0000148Z, correlation-id=8c29f6a3-2e9f-4f78-9f87-a47d2bbd83e3, azure_raw_partition_id=1, iothub-connection-auth-method={"scope":"device","type":"sas","issuer":"iothub","acceptingIpFilterRule":null}, iothub-enqueuedtime=Wed Nov 09 11:39:37 EST 2022, x-opt-enqueued-time=2022-11-09T16:39:37.387Z, iothub-connection-device-id=Biewer-Winona-Sawmill-TrimExpert-Trimmer, iothub-connection-auth-generation-id=637931467137116750, x-opt-sequence-number=107824543, iothub-message-source=Telemetry, message-id=6ba65548-2239-490a-814e-cd81a42bf676, azure_eventhubs_enqueued_time=2022-11-09T16:39:37.387Z, content-type=UTF-8, x-opt-offset=16466915418008, id=701ca58d-b94c-e766-9798-e97bb9e12995, contentType=application/json, azure_eventhubs_offset=16466915418008, timestamp=1668017171433}]
13:06:11.477 [partition-pump-1-3] INFO c.m.a.kusto.ingest.ExponentialRetry - execute: Attempt 0
.....
13:06:40.499 [partition-pump-1-3] ERROR c.a.s.i.e.i.EventHubsInboundChannelAdapter - Error occurred on partition: 1. Error: {}
com.azure.core.amqp.exception.AmqpException: New receiver 'P1-b177db87-de2c-481f-8107-f9262202a6a6' with higher epoch of '0' is created hence current receiver '77fa0ca9-d66e-41de-a787-aeb4452de72f' with epoch '0' is getting disconnected. If you are recreating the receiver, make sure a higher epoch is used. TrackingId:022626fa00000e1500080c58636bec13_G15_B5, SystemTracker:evdevcacehub:eventhub:event-hub-adx~32766|$default, Timestamp:2022-11-09T18:06:40, errorContext[NAMESPACE: evdevcacehub.servicebus.windows.net. ERROR CONTEXT: N/A, PATH: event-hub-adx/ConsumerGroups/$default/Partitions/1, REFERENCE_ID: 1_3180b5_1668017171079, LINK_CREDIT: 0]
at com.azure.core.amqp.implementation.ExceptionUtil.toException(ExceptionUtil.java:85)
at com.azure.core.amqp.implementation.handler.LinkHandler.handleRemoteLinkClosed(LinkHandler.java:120)
at com.azure.core.amqp.implementation.handler.LinkHandler.onLinkRemoteClose(LinkHandler.java:64)
at com.azure.core.amqp.implementation.handler.ReceiveLinkHandler.onLinkRemoteClose(ReceiveLinkHandler.java:228)
at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:176)
at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:91)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
13:07:10.270 [partition-pump-0-8] ERROR c.a.s.i.e.i.EventHubsInboundChannelAdapter - Error occurred on partition: 0. Error: {}
com.azure.core.amqp.exception.AmqpException: New receiver 'P0-b177db87-de2c-481f-8107-f9262202a6a6' with higher epoch of '0' is created hence current receiver 'c8a6784e-f0f0-4453-a6bb-1219a3412357' with epoch '0' is getting disconnected. If you are recreating the receiver, make sure a higher epoch is used. TrackingId:0ec4e3ce00000e16001ca77c636bec30_G30_B25, SystemTracker:evdevcacehub:eventhub:event-hub-adx~16383|$default, Timestamp:2022-11-09T18:07:10, errorContext[NAMESPACE: evdevcacehub.servicebus.windows.net. ERROR CONTEXT: N/A, PATH: event-hub-adx/ConsumerGroups/$default/Partitions/0, REFERENCE_ID: 0_8954b7_1668017200632, LINK_CREDIT: 500]
at com.azure.core.amqp.implementation.ExceptionUtil.toException(ExceptionUtil.java:85)
at com.azure.core.amqp.implementation.handler.LinkHandler.handleRemoteLinkClosed(LinkHandler.java:120)
at com.azure.core.amqp.implementation.handler.LinkHandler.onLinkRemoteClose(LinkHandler.java:64)
at com.azure.core.amqp.implementation.handler.ReceiveLinkHandler.onLinkRemoteClose(ReceiveLinkHandler.java:228)
at org.apache.qpid.proton.engine.BaseHandler.handle(BaseHandler.java:176)
at org.apache.qpid.proton.engine.impl.EventImpl.dispatch(EventImpl.java:108)
at org.apache.qpid.proton.reactor.impl.ReactorImpl.dispatch(ReactorImpl.java:324)
at org.apache.qpid.proton.reactor.impl.ReactorImpl.process(ReactorImpl.java:291)
at com.azure.core.amqp.implementation.ReactorExecutor.run(ReactorExecutor.java:91)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:68)
at reactor.core.scheduler.SchedulerTask.call(SchedulerTask.java:28)
at java.base/java.util.concurrent.FutureTask.run(FutureTask.java:264)
at java.base/java.util.concurrent.ScheduledThreadPoolExecutor$ScheduledFutureTask.run(ScheduledThreadPoolExecutor.java:304)
at java.base/java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1128)
at java.base/java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:628)
at java.base/java.lang.Thread.run(Thread.java:829)
13:08:17.063 [Thread-3] INFO c.m.a.kusto.data.HttpClientFactory - Closing HTTP client
13:08:17.063 [Thread-1] INFO c.m.a.kusto.data.HttpClientFactory - Closing HTTP client
13:08:17.117 [SpringApplicationShutdownHook] INFO o.s.c.s.binder.BinderErrorChannel - Channel 'application.event-hub-adx.$Default.errors' has 1 subscriber(s).
13:08:17.119 [SpringApplicationShutdownHook] INFO o.s.i.endpoint.EventDrivenConsumer - Removing {logging-channel-adapter:_org.springframework.integration.errorLogger} as a subscriber to the 'errorChannel' channel
13:08:17.119 [SpringApplicationShutdownHook] INFO o.s.i.c.PublishSubscribeChannel - Channel 'application.errorChannel' has 0 subscriber(s).
13:08:17.119 [SpringApplicationShutdownHook] INFO o.s.i.endpoint.EventDrivenConsumer - stopped bean '_org.springframework.integration.errorLogger'
13:08:17.119 [SpringApplicationShutdownHook] INFO o.s.i.e.ReactiveStreamsConsumer - stopped bean 'produce_integrationflow.org.springframework.integration.config.ConsumerEndpointFactoryBean#0'
Process finished with exit code 130
here my pom.xml
<?xml version="1.0" encoding="UTF-8"?>
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-parent</artifactId>
<version>2.7.3</version>
<relativePath/>
</parent>
<groupId>com.comact</groupId>
<artifactId>iot-to-adx</artifactId>
<version>0.0.1-SNAPSHOT</version>
<name>iot-to-adx</name>
<description>Export IOT Events to DAX</description>
<properties>
<java.version>11</java.version>
<spring-cloud-azure.version>4.4.0</spring-cloud-azure.version>
<springdoc.version>1.6.12</springdoc.version>
</properties>
<dependencies>
<dependency>
<groupId>com.microsoft.azure.kusto</groupId>
<artifactId>kusto-data</artifactId>
<version>3.2.0</version>
<exclusions>
<exclusion>
<artifactId>slf4j-simple</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>com.microsoft.azure.kusto</groupId>
<artifactId>kusto-ingest</artifactId>
<version>3.2.0</version>
<exclusions>
<exclusion>
<artifactId>slf4j-simple</artifactId>
<groupId>org.slf4j</groupId>
</exclusion>
</exclusions>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-webflux</artifactId>
</dependency>
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-starter</artifactId>
</dependency>
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-starter-actuator</artifactId>
</dependency>
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-stream-binder-eventhubs</artifactId>
<version>4.4.0</version>
</dependency>
<dependency>
<groupId>org.projectlombok</groupId>
<artifactId>lombok</artifactId>
<optional>true</optional>
</dependency>
<!-- https://mvnrepository.com/artifact/org.apache.commons/commons-lang3 -->
<dependency>
<groupId>org.apache.commons</groupId>
<artifactId>commons-lang3</artifactId>
<version>3.12.0</version>
</dependency>
<!-- https://mvnrepository.com/artifact/commons-beanutils/commons-beanutils -->
<dependency>
<groupId>commons-beanutils</groupId>
<artifactId>commons-beanutils</artifactId>
<version>1.9.4</version>
</dependency>
<dependency>
<groupId>org.springdoc</groupId>
<artifactId>springdoc-openapi-webflux-ui</artifactId>
<version>${springdoc.version}</version>
</dependency>
<dependency>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-starter-test</artifactId>
<scope>test</scope>
</dependency>
<dependency>
<groupId>io.projectreactor</groupId>
<artifactId>reactor-test</artifactId>
<scope>test</scope>
</dependency>
</dependencies>
<dependencyManagement>
<dependencies>
<dependency>
<groupId>com.azure.spring</groupId>
<artifactId>spring-cloud-azure-dependencies</artifactId>
<version>${spring-cloud-azure.version}</version>
<type>pom</type>
<scope>import</scope>
</dependency>
</dependencies>
</dependencyManagement>
<build>
<finalName>iot-to-adx</finalName>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-compiler-plugin</artifactId>
<version>3.3</version>
<configuration>
<source>${java.version}</source>
<target>${java.version}</target>
</configuration>
</plugin>
<plugin>
<groupId>org.springframework.boot</groupId>
<artifactId>spring-boot-maven-plugin</artifactId>
</plugin>
<plugin>
<groupId>org.springdoc</groupId>
<artifactId>springdoc-openapi-maven-plugin</artifactId>
<version>1.4</version>
</plugin>
</plugins>
</build>
</project>
my application.yaml for the Spring boot application
management:
endpoints:
web:
exposure:
include: "*"
spring:
cloud:
azure:
eventhubs:
connection-string: 'Endpoint=sb://<REMOVED>/;SharedAccessKeyName=RootManageSharedAccessKey;SharedAccessKey=<REMOVED>'
processor:
checkpoint-store:
account-name: '<REMOVED>'
account-key: '<REMOVED>'
container-name: '<REMOVED>'
stream:
function:
definition: devicemessages;produce;
bindings:
devicemessages-in-0:
destination: 'event-hub-adx'
group: '$Default'
produce-out-0:
destination: 'event-hub-adx-ingestion-errors'
dataexplorer:
hostName: '<REMOVED>'
appId: "<REMOVED>"
appKey: "<REMOVED>"
appTenant: "<REMOVED>"
kpi:
table: "kpi"
mappingTable: "kpi_mapping"
...
springdoc:
version: @springdoc.version@
show-actuator: true
use-management-port: false
My Consumer
package com.comact.iothub.consumer;
...
import com.comact.iothub.message.HubMessage;
import com.comact.iothub.message.KpiMessage;
import com.comact.iothub.message.TelemetryMessage;
import com.comact.iothub.model.Kpi;
import com.comact.iothub.producer.ErrorMessageProducer;
import com.comact.iothub.service.DataExplorerService;
import com.fasterxml.jackson.databind.ObjectMapper;
import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.exception.ExceptionUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.messaging.Message;
import java.util.function.Consumer;
@Configuration
public class TelemetryMessageConsumer {
private static final Logger LOGGER = LoggerFactory.getLogger(TelemetryMessageConsumer.class);
@Autowired
ObjectMapper objectMapper;
@Autowired
DataExplorerService dataExplorerService;
@Autowired
ErrorMessageProducer errorMessageProducer;
@Bean
Consumer<Message<String>> devicemessages() {
return message -> {
LOGGER.info("Consumed a new message [{}] with headers [{}]", message.getPayload(), message.getHeaders());
try {
...
// need to find the right Model to load from the payload
var telemetryMessage = objectMapper.readValue(message.getPayload(), TelemetryMessage.class);
switch (telemetryMessage.telemetryType){
case "KPI":
LOGGER.trace("KPI message received");
var kpiMessage = objectMapper.readValue(message.getPayload(), KpiMessage.class);
var kpi = Kpi.builder()
.kpiMessage(kpiMessage)
.hubMessage(hubMessage)
.build();
dataExplorerService.ingestKpi(kpi);
break;
default:
LOGGER.error("UNKNOWN MESSAGE TYPE RECEIVED [{}] headers [{}]", message.getPayload(), message.getHeaders());
var errorMessage = new ErrorMessage();
errorMessage.setErrorMessage("UNKNOWN MESSAGE TYPE RECEIVED");
errorMessage.setMessagePayload(message.getPayload());
errorMessage.setMessageHeaders(headers);
errorMessageProducer.sendMessage(errorMessage);
}
} catch (Exception e) {
LOGGER.error("There are errors while processing a message", e);
var errorMessage = new ErrorMessage();
errorMessage.setErrorMessage(String.format("There are errors while processing a message exception [%s]", ExceptionUtils.getStackTrace(e)));
errorMessage.setMessagePayload(message.getPayload());
try {
errorMessage.setMessageHeaders(objectMapper.writeValueAsString(message.getHeaders()));
} catch (Exception ignored){}
errorMessageProducer.sendMessage(errorMessage);
}
};
}
}
How I ingest data into ADX
package com.comact.iothub.service;
import com.comact.iothub.model.Downtime;
import com.comact.iothub.model.Shift;
import com.comact.iothub.model.Kpi;
import com.comact.iothub.model.TelemetryModel;
import com.comact.iothub.producer.ErrorMessageProducer;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
import com.microsoft.azure.kusto.ingest.IngestClientFactory;
import com.microsoft.azure.kusto.ingest.IngestionMapping;
import com.microsoft.azure.kusto.ingest.IngestionProperties;
import com.microsoft.azure.kusto.ingest.ManagedStreamingIngestClient;
import com.microsoft.azure.kusto.ingest.result.IngestionResult;
import com.microsoft.azure.kusto.ingest.source.StreamSourceInfo;
import org.apache.commons.lang3.StringUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Service;
import javax.annotation.PostConstruct;
import java.io.ByteArrayInputStream;
import java.nio.charset.StandardCharsets;
@Service
public class DataExplorerService {
private static final Logger LOGGER = LoggerFactory.getLogger(DataExplorerService.class);
@Autowired
ObjectMapper objectMapper;
@Autowired
ErrorMessageProducer errorMessageProducer;
ManagedStreamingIngestClient streamingIngestClient;
@Value("${dataexplorer.hostName}")
private String hostName;
@Value("${dataexplorer.appId}")
private String appId;
@Value("${dataexplorer.appKey}")
private String appKey;
@Value("${dataexplorer.appTenant}")
private String appTenant;
@Value("${dataexplorer.kpi.table}")
private String kpiTable;
@Value("${dataexplorer.kpi.mappingTable}")
private String kpiMappingTable;
@PostConstruct
private void init() throws Exception {
var csb = ConnectionStringBuilder.createWithAadApplicationCredentials(hostName, appId, appKey, appTenant);
streamingIngestClient = IngestClientFactory.createManagedStreamingIngestClient(csb);
}
public IngestionResult ingestKpi(Kpi model) throws Exception {
var table = kpiTable;
var mapping = kpiMappingTable;
var serializedModel = objectMapper.writeValueAsString(model);
var ingestionResult = ingest(getDatabaseName(model), serializedModel, table, mapping);
return ingestionResult;
}
private IngestionResult ingest(String database, String serializedModel, String table, String mapping) throws Exception {
var ingestionProperties = new IngestionProperties(database, table);
ingestionProperties.setDataFormat(IngestionProperties.DataFormat.JSON);
ingestionProperties.setIngestionMapping(mapping, IngestionMapping.IngestionMappingKind.JSON);
LOGGER.debug("Ingesting data in database [{}] / table [{}] with mapping table [{}]", database, table, mapping);
var inputStream = new ByteArrayInputStream(StandardCharsets.UTF_8.encode(serializedModel).array());
var streamSourceInfo = new StreamSourceInfo(inputStream);
return streamingIngestClient.ingestFromStream(streamSourceInfo, ingestionProperties);
}
private String getDatabaseName(TelemetryModel model){
var name = .....
// Database name can include only alphanumeric, spaces, dash, underscore, and dot characters, and be up to 260 characters in length.
name = name.replaceAll("[^a-zA-z-0-9._ ]","");
return StringUtils.truncate(name, 260);
}
}
The eventhub for incoming messages are only used by this application and I’m using the $Default consumer-group.
I don’t know what is happening.
Setup (please complete the following information):
- OS: Windows 10
- IDE: [IntelliJ]
- Library/Libraries: see pom.xml
- Java version: [11]
- App Server/Environment: [
-
- Spring Apps Service
- Spring App
- EventHub Namespace
- EventHub (like topics)
- Storage Account for EventHub
- Storage Container for EventHub
- EventHub Consumer group
- IotHub
- IotHub route Endpoints
- IotHub routes
- Data Explorer (with databases and tables)
- Application Registration]
- Frameworks: [Spring Boot]
Information Checklist Kindly make sure that you have added all the following information above and checkoff the required fields otherwise we will treat the issuer as an incomplete report
- Bug Description Added
- [ X] Repro Steps Added
- [ X] Setup information Added
About this issue
- Original URL
- State: closed
- Created 2 years ago
- Comments: 28 (14 by maintainers)
Hi Zeija, could you help take a look at the logs, please? \cc @ZejiaJiang
@yiliuTo here the logs. I replace some payload (because the data is confidential) debug.log
Thank you @survivant !
I think this issue is solved. I’m going to close this issue. if you meet same issue again, you can reopen it or create a new issue.
@survivant thanks for reaching out,
@yiliuTo could you help handle this issue?