You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by lh...@apache.org on 2024/02/13 12:41:34 UTC
(pulsar) branch master updated: [feat][io] AzureDataExplorer/Kusto Sink for Pulsar (#22006)
This is an automated email from the ASF dual-hosted git repository.
lhotari pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/master by this push:
new beed0cfc52e [feat][io] AzureDataExplorer/Kusto Sink for Pulsar (#22006)
beed0cfc52e is described below
commit beed0cfc52e566f3a5293f7ea37fdcc6e334b0f0
Author: Abhishek Saharn <10...@users.noreply.github.com>
AuthorDate: Tue Feb 13 18:11:27 2024 +0530
[feat][io] AzureDataExplorer/Kusto Sink for Pulsar (#22006)
- https://learn.microsoft.com/en-us/azure/data-explorer/
Co-authored-by: Ramachandran A G <10...@users.noreply.github.com>
---
distribution/io/src/assemble/io.xml | 1 +
pulsar-io/azure-data-explorer/docker-compose.yml | 47 ++++
pulsar-io/azure-data-explorer/pom.xml | 83 ++++++
.../io/azuredataexplorer/ADXPulsarEvent.java | 34 +++
.../pulsar/io/azuredataexplorer/ADXSink.java | 284 +++++++++++++++++++++
.../pulsar/io/azuredataexplorer/ADXSinkConfig.java | 105 ++++++++
.../pulsar/io/azuredataexplorer/ADXSinkUtils.java | 41 +++
.../pulsar/io/azuredataexplorer/package-info.java | 19 ++
.../resources/META-INF/services/pulsar-io.yaml | 23 ++
.../io/azuredataexplorer/ADXSinkConfigTest.java | 113 ++++++++
.../io/azuredataexplorer/ADXSinkE2ETest.java | 175 +++++++++++++
.../src/test/resources/sinkConfig.yaml | 33 +++
.../src/test/resources/sinkConfigInvalid.yaml | 26 ++
.../src/test/resources/sinkConfigValid.yaml | 33 +++
pulsar-io/docs/pom.xml | 5 +
pulsar-io/pom.xml | 4 +-
16 files changed, 1025 insertions(+), 1 deletion(-)
diff --git a/distribution/io/src/assemble/io.xml b/distribution/io/src/assemble/io.xml
index 5b652170fdb..f98ee14bb20 100644
--- a/distribution/io/src/assemble/io.xml
+++ b/distribution/io/src/assemble/io.xml
@@ -81,5 +81,6 @@
<file><source>${basedir}/../../pulsar-io/solr/target/pulsar-io-solr-${project.version}.nar</source></file>
<file><source>${basedir}/../../pulsar-io/dynamodb/target/pulsar-io-dynamodb-${project.version}.nar</source></file>
<file><source>${basedir}/../../pulsar-io/alluxio/target/pulsar-io-alluxio-${project.version}.nar</source></file>
+ <file><source>${basedir}/../../pulsar-io/azure-data-explorer/target/pulsar-io-azuredataexplorer-${project.version}.nar</source></file>
</files>
</assembly>
diff --git a/pulsar-io/azure-data-explorer/docker-compose.yml b/pulsar-io/azure-data-explorer/docker-compose.yml
new file mode 100644
index 00000000000..4825d92edeb
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/docker-compose.yml
@@ -0,0 +1,47 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+
+version: '3.0'
+
+services:
+ pulsar-server:
+ command: bin/pulsar standalone
+ image: apachepulsar/pulsar-all:latest
+ container_name: pulsar-server
+ hostname: pulsar-server
+ volumes:
+ - ./target/pulsar-io-azuredataexplorer-3.2.0-SNAPSHOT.nar:/pulsar/connectors/pulsar-io-azuredataexplorer-3.2.0-SNAPSHOT.nar
+ ports:
+ - 8080:8080
+ - 6650:6650
+ networks:
+ - custom-sink-connector
+ healthcheck:
+ interval: 10s
+ retries: 20
+ test: curl --write-out 'HTTP %{http_code}' --fail --silent --output /dev/null http://localhost:8080/admin/v2/clusters/standalone
+
+volumes:
+ custom-sink-connector:
+ driver: local
+
+networks:
+ custom-sink-connector:
+ driver: bridge
\ No newline at end of file
diff --git a/pulsar-io/azure-data-explorer/pom.xml b/pulsar-io/azure-data-explorer/pom.xml
new file mode 100644
index 00000000000..5e855c74714
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/pom.xml
@@ -0,0 +1,83 @@
+<!--
+
+ Licensed to the Apache Software Foundation (ASF) under one
+ or more contributor license agreements. See the NOTICE file
+ distributed with this work for additional information
+ regarding copyright ownership. The ASF licenses this file
+ to you under the Apache License, Version 2.0 (the
+ "License"); you may not use this file except in compliance
+ with the License. You may obtain a copy of the License at
+
+ http://www.apache.org/licenses/LICENSE-2.0
+
+ Unless required by applicable law or agreed to in writing,
+ software distributed under the License is distributed on an
+ "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ KIND, either express or implied. See the License for the
+ specific language governing permissions and limitations
+ under the License.
+
+-->
+<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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+ <modelVersion>4.0.0</modelVersion>
+ <parent>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-io</artifactId>
+ <version>3.3.0-SNAPSHOT</version>
+ </parent>
+
+ <artifactId>pulsar-io-azuredataexplorer</artifactId>
+ <name>Pulsar IO :: AzureDataExplorer</name>
+
+ <properties>
+ <kusto.sdk.version>5.0.4</kusto.sdk.version>
+ </properties>
+ <dependencies>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-core</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.pulsar</groupId>
+ <artifactId>pulsar-functions-instance</artifactId>
+ <version>${project.version}</version>
+ <scope>test</scope>
+ </dependency>
+ <!-- https://mvnrepository.com/artifact/com.microsoft.azure.kusto/kusto-data -->
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-data</artifactId>
+ <version>${kusto.sdk.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.datatype</groupId>
+ <artifactId>jackson-datatype-jsr310</artifactId>
+ </dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.dataformat</groupId>
+ <artifactId>jackson-dataformat-yaml</artifactId>
+ </dependency>
+ <!-- https://mvnrepository.com/artifact/com.microsoft.azure.kusto/kusto-ingest -->
+ <dependency>
+ <groupId>com.microsoft.azure.kusto</groupId>
+ <artifactId>kusto-ingest</artifactId>
+ <version>${kusto.sdk.version}</version>
+ </dependency>
+ </dependencies>
+ <build>
+ <plugins>
+ <plugin>
+ <groupId>org.apache.nifi</groupId>
+ <artifactId>nifi-nar-maven-plugin</artifactId>
+ </plugin>
+ </plugins>
+ </build>
+</project>
\ No newline at end of file
diff --git a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXPulsarEvent.java b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXPulsarEvent.java
new file mode 100644
index 00000000000..3c64fc7be06
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXPulsarEvent.java
@@ -0,0 +1,34 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
+
+import java.time.Instant;
+import lombok.Data;
+import lombok.experimental.Accessors;
+
+@Data
+@Accessors(chain = true)
+public class ADXPulsarEvent {
+ private String key;
+ private String value;
+ private String properties;
+ private String producerName;
+ private long sequenceId;
+ private Instant eventTime;
+}
diff --git a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java
new file mode 100644
index 00000000000..07eb372833b
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSink.java
@@ -0,0 +1,284 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.datatype.jsr310.JavaTimeModule;
+import com.microsoft.azure.kusto.data.StringUtils;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import com.microsoft.azure.kusto.data.exceptions.KustoDataExceptionBase;
+import com.microsoft.azure.kusto.ingest.IngestClient;
+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.exceptions.IngestionClientException;
+import com.microsoft.azure.kusto.ingest.exceptions.IngestionServiceException;
+import com.microsoft.azure.kusto.ingest.result.IngestionResult;
+import com.microsoft.azure.kusto.ingest.result.IngestionStatus;
+import com.microsoft.azure.kusto.ingest.result.TableReportIngestionResult;
+import com.microsoft.azure.kusto.ingest.source.StreamSourceInfo;
+import java.io.ByteArrayInputStream;
+import java.net.ConnectException;
+import java.net.URISyntaxException;
+import java.nio.charset.StandardCharsets;
+import java.time.Instant;
+import java.util.ArrayList;
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Map;
+import java.util.UUID;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.client.api.PulsarClientException;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.io.core.Sink;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.core.annotations.Connector;
+import org.apache.pulsar.io.core.annotations.IOType;
+import org.jetbrains.annotations.NotNull;
+
+@Connector(
+ name = "adx",
+ type = IOType.SINK,
+ help = "The ADXSink is used for moving messages from Pulsar to ADX.",
+ configClass = ADXSinkConfig.class
+)
+@Slf4j
+public class ADXSink implements Sink<byte[]> {
+ private final ObjectMapper mapper = new ObjectMapper().registerModule(new JavaTimeModule());
+ IngestionProperties ingestionProperties;
+ private IngestClient ingestClient;
+ private List<Record<byte[]>> incomingRecordsList;
+ private int batchSize;
+ private long batchTimeMs;
+ private ScheduledExecutorService adxSinkExecutor;
+ private int maxRetryAttempts;
+ private long retryBackOffTime;
+
+ @Override
+ public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
+ log.info("Open ADX Sink");
+ // Azure data explorer, initializations
+ ADXSinkConfig adxConfig = ADXSinkConfig.load(config, sinkContext);
+ adxConfig.validate();
+ ConnectionStringBuilder kcsb = getConnectionStringBuilder(adxConfig);
+ if (kcsb == null) {
+ throw new Exception("Kusto Connection String NULL");
+ }
+ log.debug("ConnectionString created: {}.", kcsb);
+ ingestClient = adxConfig.getManagedIdentityId() != null
+ ? IngestClientFactory.createManagedStreamingIngestClient(kcsb) :
+ IngestClientFactory.createClient(kcsb);
+ ingestionProperties = new IngestionProperties(adxConfig.getDatabase(), adxConfig.getTable());
+ ingestionProperties.setIngestionMapping(adxConfig.getMappingRefName(),
+ getParseMappingRefType(adxConfig.getMappingRefType()));
+ ingestionProperties.setReportLevel(IngestionProperties.IngestionReportLevel.FAILURES_AND_SUCCESSES);
+ ingestionProperties.setReportMethod(IngestionProperties.IngestionReportMethod.TABLE);
+ ingestionProperties.setFlushImmediately(adxConfig.isFlushImmediately());
+ ingestionProperties.setDataFormat(IngestionProperties.DataFormat.MULTIJSON);
+ log.debug("Ingestion Properties: {}", ingestionProperties.toString());
+
+ maxRetryAttempts = adxConfig.getMaxRetryAttempts() + 1;
+ retryBackOffTime = adxConfig.getRetryBackOffTime();
+ /*incoming records list will hold incoming messages,
+ flushExecutor executes the flushData according to batch time */
+ batchSize = adxConfig.getBatchSize();
+ batchTimeMs = adxConfig.getBatchTimeMs();
+ incomingRecordsList = new ArrayList<>();
+ adxSinkExecutor = Executors.newScheduledThreadPool(1);
+ adxSinkExecutor.scheduleAtFixedRate(this::sinkData, batchTimeMs, batchTimeMs, TimeUnit.MILLISECONDS);
+ }
+
+ @Override
+ public void write(Record<byte[]> record) {
+ int runningSize;
+ synchronized (this) {
+ incomingRecordsList.add(record);
+ runningSize = incomingRecordsList.size();
+ }
+ if (runningSize == batchSize) {
+ adxSinkExecutor.execute(this::sinkData);
+ }
+ }
+
+ private void sinkData() {
+ final List<Record<byte[]>> recordsToSink;
+ synchronized (this) {
+ if (incomingRecordsList.isEmpty()) {
+ return;
+ }
+ recordsToSink = incomingRecordsList;
+ incomingRecordsList = new ArrayList<>();
+ }
+
+ List<ADXPulsarEvent> eventsToSink = new LinkedList<>();
+ for (Record<byte[]> record : recordsToSink) {
+ try {
+ eventsToSink.add(getADXPulsarEvent(record));
+ } catch (Exception ex) {
+ record.fail();
+ log.error("Failed to collect the record for ADX cluster.", ex);
+ }
+ }
+ try {
+ for (int retryAttempts = 0; true; retryAttempts++) {
+ try {
+ StreamSourceInfo streamSourceInfo =
+ new StreamSourceInfo(new ByteArrayInputStream(mapper.writeValueAsBytes(eventsToSink)));
+ IngestionResult ingestionResult =
+ ingestClient.ingestFromStream(streamSourceInfo, ingestionProperties);
+ if (ingestionResult instanceof TableReportIngestionResult) {
+ // If TableReportIngestionResult returned then the ingestion status is from streaming ingest
+ IngestionStatus ingestionStatus = ingestionResult.getIngestionStatusCollection().get(0);
+ if (!hasStreamingSucceeded(ingestionStatus)) {
+ retryAttempts += ManagedStreamingIngestClient.ATTEMPT_COUNT;
+ backOffForRemainingAttempts(retryAttempts, null, recordsToSink);
+ continue;
+ }
+ recordsToSink.forEach(Record::ack);
+ }
+ return;
+ } catch (IngestionServiceException exception) {
+ Throwable innerException = exception.getCause();
+ if (innerException instanceof KustoDataExceptionBase
+ && ((KustoDataExceptionBase) innerException).isPermanent()) {
+ recordsToSink.forEach(Record::fail);
+ throw new ConnectException(exception.getMessage());
+ }
+ // retrying transient exceptions
+ backOffForRemainingAttempts(retryAttempts, exception, recordsToSink);
+ } catch (IngestionClientException | URISyntaxException exception) {
+ recordsToSink.forEach(Record::fail);
+ throw new ConnectException(exception.getMessage());
+ }
+ }
+
+ } catch (Exception ex) {
+ log.error("Failed to publish the message to ADX cluster", ex);
+ }
+ }
+
+ private boolean hasStreamingSucceeded(@NotNull IngestionStatus status) {
+ switch (status.status) {
+ case Succeeded:
+ case Queued:
+ case Pending:
+ return true;
+ case Skipped:
+ case PartiallySucceeded:
+ String failureStatus = status.getFailureStatus();
+ String details = status.getDetails();
+ UUID ingestionSourceId = status.getIngestionSourceId();
+ log.warn("A batch of streaming records has {} ingestion: table:{}, database:{}, operationId: {},"
+ + "ingestionSourceId: {}{}{}.\n"
+ + "Status is final and therefore ingestion won't be retried and data won't reach dlq",
+ status.getStatus(),
+ status.getTable(),
+ status.getDatabase(),
+ status.getOperationId(),
+ ingestionSourceId,
+ (StringUtils.isNotEmpty(failureStatus) ? (", failure: " + failureStatus) : ""),
+ (StringUtils.isNotEmpty(details) ? (", details: " + details) : ""));
+ return true;
+ case Failed:
+ }
+ return false;
+ }
+
+ private void backOffForRemainingAttempts(int retryAttempts, Exception exception, List<Record<byte[]>> records)
+ throws PulsarClientException.ConnectException {
+ if (retryAttempts < maxRetryAttempts) {
+ long sleepTimeMs = retryBackOffTime;
+ log.error(
+ "Failed to ingest records into Kusto, backing off and retrying ingesting records "
+ + "after {} milliseconds.",
+ sleepTimeMs);
+ try {
+ TimeUnit.MILLISECONDS.sleep(sleepTimeMs);
+ throw new InterruptedException();
+ } catch (InterruptedException interruptedErr) {
+ records.forEach(Record::fail);
+ throw new PulsarClientException.ConnectException(String.format(
+ "Retrying ingesting records into KustoDB was interrupted after retryAttempts=%s",
+ retryAttempts + 1)
+ );
+ }
+ } else {
+ records.forEach(Record::fail);
+ throw new PulsarClientException.ConnectException(
+ String.format("Retry attempts exhausted, failed to ingest records into KustoDB. Exception: %s",
+ exception.getMessage()));
+ }
+ }
+
+ private @NotNull ADXPulsarEvent getADXPulsarEvent(@NotNull Record<byte[]> record) throws Exception {
+ ADXPulsarEvent event = new ADXPulsarEvent();
+ record.getEventTime().ifPresent(time -> event.setEventTime(Instant.ofEpochMilli(time)));
+ record.getKey().ifPresent(event::setKey);
+ record.getMessage().ifPresent(message -> event.setProducerName(message.getProducerName()));
+ record.getMessage().ifPresent(message -> event.setSequenceId(message.getSequenceId()));
+ event.setValue(new String(record.getValue(), StandardCharsets.UTF_8));
+ event.setProperties(new ObjectMapper().writeValueAsString(record.getProperties()));
+ return event;
+ }
+
+ private IngestionMapping.IngestionMappingKind getParseMappingRefType(String mappingRefType) {
+ if (mappingRefType == null || mappingRefType.isEmpty()) {
+ return null;
+ }
+ return switch (mappingRefType) {
+ case "CSV" -> IngestionMapping.IngestionMappingKind.CSV;
+ case "AVRO" -> IngestionMapping.IngestionMappingKind.AVRO;
+ case "JSON" -> IngestionMapping.IngestionMappingKind.JSON;
+ case "PARQUET" -> IngestionMapping.IngestionMappingKind.PARQUET;
+ default -> null;
+ };
+ }
+
+ private ConnectionStringBuilder getConnectionStringBuilder(@NotNull ADXSinkConfig adxConfig) {
+
+ if (adxConfig.getManagedIdentityId() != null) {
+ if ("system".equalsIgnoreCase(adxConfig.getManagedIdentityId())) {
+ return ConnectionStringBuilder.createWithAadManagedIdentity(adxConfig.getClusterUrl());
+ }
+ ConnectionStringBuilder.createWithAadManagedIdentity(adxConfig.getClusterUrl(),
+ adxConfig.getManagedIdentityId());
+ }
+ return ConnectionStringBuilder.createWithAadApplicationCredentials(adxConfig.getClusterUrl(),
+ adxConfig.getAppId(), adxConfig.getAppKey(), adxConfig.getTenantId());
+ }
+
+ @Override
+ public void close() throws Exception {
+ ingestClient.close();
+ adxSinkExecutor.shutdown();
+ try {
+ if (!adxSinkExecutor.awaitTermination(2 * batchTimeMs, TimeUnit.MILLISECONDS)) {
+ adxSinkExecutor.shutdownNow();
+ }
+ } catch (InterruptedException ie) {
+ adxSinkExecutor.shutdownNow();
+ Thread.currentThread().interrupt();
+ }
+ log.info("Kusto ingest client closed.");
+ }
+}
\ No newline at end of file
diff --git a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfig.java b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfig.java
new file mode 100644
index 00000000000..42c583bd3cc
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfig.java
@@ -0,0 +1,105 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
+
+import com.fasterxml.jackson.databind.ObjectMapper;
+import com.fasterxml.jackson.dataformat.yaml.YAMLFactory;
+import java.io.File;
+import java.io.IOException;
+import java.io.Serializable;
+import java.util.Map;
+import java.util.Objects;
+import lombok.Data;
+import lombok.experimental.Accessors;
+import org.apache.pulsar.io.common.IOConfigUtils;
+import org.apache.pulsar.io.core.SinkContext;
+import org.apache.pulsar.io.core.annotations.FieldDoc;
+
+
+@Data
+@Accessors(chain = true)
+public class ADXSinkConfig implements Serializable {
+
+ @FieldDoc(required = true, defaultValue = "", help = "The ADX cluster URL")
+ private String clusterUrl;
+
+ @FieldDoc(required = true, defaultValue = "", help = "The database name to which data need to be ingested")
+ private String database;
+
+ @FieldDoc(required = true, defaultValue = "", help = "Table name to which pulsar data need to be ingested.")
+ private String table;
+
+ @FieldDoc(defaultValue = "", help = "The AAD app Id for authentication", sensitive = true)
+ private String appId;
+
+ @FieldDoc(defaultValue = "", help = "The AAD app secret for authentication", sensitive = true)
+ private String appKey;
+
+ @FieldDoc(defaultValue = "", help = "The tenant Id for authentication")
+ private String tenantId;
+
+ @FieldDoc(defaultValue = "", help = "The Managed Identity credential for authentication."
+ + " Set this with clientId in case of User assigned MI."
+ + " and 'system' in case of System assigned managed identity")
+ private String managedIdentityId;
+
+ @FieldDoc(defaultValue = "", help = "The mapping reference for ingestion")
+ private String mappingRefName;
+
+ @FieldDoc(defaultValue = "", help = "The type of mapping reference provided")
+ private String mappingRefType;
+
+ @FieldDoc(defaultValue = "false", help = "Denotes if flush should happen immediately without aggregation. "
+ + "Not recommended to enable flushImmediately for production workloads")
+ private boolean flushImmediately = false;
+
+ @FieldDoc(defaultValue = "100", help = "For batching, this defines the number of "
+ + "records to hold for batching, to sink data to adx")
+ private int batchSize = 100;
+
+ @FieldDoc(defaultValue = "10000", help = "For batching, this defines the time to hold"
+ + " records before sink to adx")
+ private long batchTimeMs = 10000;
+
+ @FieldDoc(defaultValue = "1", help = "Max retry attempts, In case of transient ingestion errors")
+ private int maxRetryAttempts = 1;
+
+ @FieldDoc(defaultValue = "10", help = "Period of time in milliseconds to backoff"
+ + " before retry for transient errors")
+ private long retryBackOffTime = 10;
+
+
+ public static ADXSinkConfig load(String yamlFile) throws IOException {
+ ObjectMapper mapper = new ObjectMapper(new YAMLFactory());
+ return mapper.readValue(new File(yamlFile), ADXSinkConfig.class);
+ }
+
+ protected static ADXSinkConfig load(Map<String, Object> config, SinkContext sinkContext) {
+ return IOConfigUtils.loadWithSecrets(config, ADXSinkConfig.class, sinkContext);
+ }
+
+ public void validate() throws Exception {
+ Objects.requireNonNull(clusterUrl, "clusterUrl property not set.");
+ Objects.requireNonNull(database, "database property not set.");
+ Objects.requireNonNull(table, "table property not set.");
+ if (managedIdentityId == null && (appId == null || appKey == null || tenantId == null)) {
+ throw new Exception("Auth credentials not valid");
+ }
+ }
+}
diff --git a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkUtils.java b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkUtils.java
new file mode 100644
index 00000000000..4a073f8bf7a
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkUtils.java
@@ -0,0 +1,41 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
+
+public class ADXSinkUtils {
+
+ static final String INGEST_PREFIX = "ingest-";
+ static final String PROTOCOL_SUFFIX = "://";
+
+ static String getIngestionEndpoint(String clusterUrl) {
+ if (clusterUrl.contains(INGEST_PREFIX)) {
+ return clusterUrl;
+ } else {
+ return clusterUrl.replaceFirst(PROTOCOL_SUFFIX, PROTOCOL_SUFFIX + INGEST_PREFIX);
+ }
+ }
+
+ static String getQueryEndpoint(String clusterUrl) {
+ if (clusterUrl.contains(INGEST_PREFIX)) {
+ return clusterUrl.replaceFirst(INGEST_PREFIX, "");
+ } else {
+ return clusterUrl;
+ }
+ }
+}
diff --git a/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/package-info.java b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/package-info.java
new file mode 100644
index 00000000000..9a9bd7d4db6
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/main/java/org/apache/pulsar/io/azuredataexplorer/package-info.java
@@ -0,0 +1,19 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
\ No newline at end of file
diff --git a/pulsar-io/azure-data-explorer/src/main/resources/META-INF/services/pulsar-io.yaml b/pulsar-io/azure-data-explorer/src/main/resources/META-INF/services/pulsar-io.yaml
new file mode 100644
index 00000000000..753bfb3ff05
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/main/resources/META-INF/services/pulsar-io.yaml
@@ -0,0 +1,23 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+name: azure-data-explorer
+description: Azure Data Explorer sink
+sinkClass: org.apache.pulsar.io.azuredataexplorer.ADXSink
+sinkConfigClass: org.apache.pulsar.io.azuredataexplorer.ADXSinkConfig
diff --git a/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java b/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java
new file mode 100644
index 00000000000..c05e3b159d8
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkConfigTest.java
@@ -0,0 +1,113 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
+
+import org.jetbrains.annotations.NotNull;
+import org.testng.annotations.Test;
+
+import java.io.File;
+import java.io.IOException;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Objects;
+import org.apache.pulsar.io.core.SinkContext;
+import org.mockito.Mockito;
+
+import static org.testng.Assert.*;
+
+public class ADXSinkConfigTest {
+ @Test
+ public final void loadFromYamlFileTest() throws IOException {
+ File sinkConfig = readSinkConfig("sinkConfig.yaml");
+ String path = sinkConfig.getAbsolutePath();
+ ADXSinkConfig config = ADXSinkConfig.load(path);
+ assertNotNull(config);
+ assertEquals(config.getClusterUrl(), "https://somecluster.eastus.kusto.windows.net");
+ assertEquals(config.getDatabase(), "somedb");
+ assertEquals(config.getTable(), "tableName");
+ assertEquals(config.getAppId(), "xxxx-xxxx-xxxx-xxxx");
+ assertEquals(config.getAppKey(), "xxxx-xxxx-xxxx-xxxx");
+ assertEquals(config.getTenantId(), "xxxx-xxxx-xxxx-xxxx");
+ assertEquals(config.getManagedIdentityId(), "xxxx-some-id-xxxx OR empty string");
+ assertEquals(config.getMappingRefName(), "mapping ref name");
+ assertEquals(config.getMappingRefType(), "CSV");
+ assertFalse(config.isFlushImmediately());
+ assertEquals(config.getBatchSize(), 100);
+ assertEquals(config.getBatchTimeMs(), 10000);
+ }
+
+ @Test
+ public final void validateConfigTest() throws Exception {
+ File yamlFile = readSinkConfig("sinkConfigValid.yaml");
+ String path = yamlFile.getAbsolutePath();
+ ADXSinkConfig config = ADXSinkConfig.load(path);
+ config.validate();
+ }
+
+ @Test(expectedExceptions = Exception.class)
+ public final void validateInvalidConfigTest() throws Exception {
+ File yamlFile = readSinkConfig("sinkConfigInvalid.yaml");
+ String path = yamlFile.getAbsolutePath();
+ ADXSinkConfig config = ADXSinkConfig.load(path);
+ config.validate();
+ }
+
+ @Test
+ public final void loadFromMapTest() throws IOException {
+ Map<String, Object> map = getConfig();
+ SinkContext sinkContext = Mockito.mock(SinkContext.class);
+ ADXSinkConfig config = ADXSinkConfig.load(map, sinkContext);
+ assertNotNull(config);
+ assertEquals(config.getClusterUrl(), "https://somecluster.eastus.kusto.windows.net");
+ assertEquals(config.getDatabase(), "somedb");
+ assertEquals(config.getTable(), "tableName");
+ assertEquals(config.getAppId(), "xxxx-xxxx-xxxx-xxxx");
+ assertEquals(config.getAppKey(), "xxxx-xxxx-xxxx-xxxx");
+ assertEquals(config.getTenantId(), "xxxx-xxxx-xxxx-xxxx");
+ assertEquals(config.getManagedIdentityId(), "xxxx-some-id-xxxx OR empty string");
+ assertEquals(config.getMappingRefName(), "mapping ref name");
+ assertEquals(config.getMappingRefType(), "CSV");
+ assertFalse(config.isFlushImmediately());
+ assertEquals(config.getBatchSize(), 100);
+ assertEquals(config.getBatchTimeMs(), 10000);
+ }
+
+ @NotNull
+ private static Map<String, Object> getConfig() {
+ Map<String, Object> map = new HashMap<>();
+ map.put("clusterUrl", "https://somecluster.eastus.kusto.windows.net");
+ map.put("database", "somedb");
+ map.put("table", "tableName");
+ map.put("appId", "xxxx-xxxx-xxxx-xxxx");
+ map.put("appKey", "xxxx-xxxx-xxxx-xxxx");
+ map.put("tenantId", "xxxx-xxxx-xxxx-xxxx");
+ map.put("managedIdentityId", "xxxx-some-id-xxxx OR empty string");
+ map.put("mappingRefName", "mapping ref name");
+ map.put("mappingRefType", "CSV");
+ map.put("flushImmediately", false);
+ map.put("batchSize", 100);
+ map.put("batchTimeMs", 10000);
+ return map;
+ }
+
+ private @NotNull File readSinkConfig(String name) {
+ ClassLoader classLoader = getClass().getClassLoader();
+ return new File(Objects.requireNonNull(classLoader.getResource(name)).getFile());
+ }
+}
diff --git a/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkE2ETest.java b/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkE2ETest.java
new file mode 100644
index 00000000000..b04c20dad89
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/test/java/org/apache/pulsar/io/azuredataexplorer/ADXSinkE2ETest.java
@@ -0,0 +1,175 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one
+ * or more contributor license agreements. See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. The ASF licenses this file
+ * to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied. See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+package org.apache.pulsar.io.azuredataexplorer;
+
+import com.microsoft.azure.kusto.data.Client;
+import com.microsoft.azure.kusto.data.ClientFactory;
+import com.microsoft.azure.kusto.data.KustoOperationResult;
+import com.microsoft.azure.kusto.data.KustoResultSetTable;
+import com.microsoft.azure.kusto.data.auth.ConnectionStringBuilder;
+import lombok.extern.slf4j.Slf4j;
+import org.apache.pulsar.functions.api.Record;
+import org.apache.pulsar.functions.instance.SinkRecord;
+import org.apache.pulsar.io.core.SinkContext;
+import org.mockito.Mockito;
+import org.testng.Assert;
+import org.testng.SkipException;
+import org.testng.annotations.AfterMethod;
+import org.testng.annotations.BeforeMethod;
+import org.testng.annotations.Test;
+
+import java.nio.charset.StandardCharsets;
+import java.util.HashMap;
+import java.util.Map;
+import java.util.Optional;
+import java.util.concurrent.ThreadLocalRandom;
+
+@Slf4j
+public class ADXSinkE2ETest {
+
+ String database;
+ String cluster;
+ String authorityId;
+ String appId;
+ String appKey;
+ private final String table = "ADXPulsarTest_" + ThreadLocalRandom.current().nextInt(0, 100);
+ private Client kustoAdminClient = null;
+ Map<String, Object> configs;
+
+ @BeforeMethod
+ public void setUp() throws Exception {
+ cluster = System.getenv("kustoCluster");
+ database = System.getenv("kustoDatabase");
+ authorityId = System.getenv("kustoAadAuthorityID");
+ appId = System.getenv("kustoAadAppId");
+ appKey = System.getenv("kustoAadAppSecret");
+
+ if(cluster == null){
+ throw new SkipException("Skipping tests because environment vars was not accessible.");
+ }
+
+ configs = new HashMap<>();
+ configs.put("clusterUrl", cluster);
+ configs.put("database", database);
+ configs.put("table", table);
+ configs.put("batchTimeMs", 1000);
+ configs.put("flushImmediately", true);
+ configs.put("appId", appId);
+ configs.put("appKey", appKey);
+ configs.put("tenantId", authorityId);
+ configs.put("maxRetryAttempts", 3);
+ configs.put("retryBackOffTime", 100);
+
+ ConnectionStringBuilder engineKcsb =
+ ConnectionStringBuilder.createWithAadApplicationCredentials(ADXSinkUtils.getQueryEndpoint(cluster),
+ appId, appKey, authorityId);
+ kustoAdminClient = ClientFactory.createClient(engineKcsb);
+ String createTableCommand = ".create table " + table +
+ " ( key:string , value:string, eventTime:datetime , producerName:string , sequenceId:long ,properties:dynamic )";
+ log.info("Creating test table {} ", table);
+ kustoAdminClient.execute(database, createTableCommand);
+ kustoAdminClient.execute(database, generateAlterIngestionBatchingPolicyCommand(database,
+ "{\"MaximumBatchingTimeSpan\":\"00:00:10\", \"MaximumNumberOfItems\": 500, \"MaximumRawDataSizeMB\": 1024}"));
+ log.info("Ingestion policy on table {} altered",table);
+ }
+
+ private String generateAlterIngestionBatchingPolicyCommand(String entityName, String targetBatchingPolicy) {
+ return ".alter database " + entityName + " policy ingestionbatching ```" + targetBatchingPolicy + "```";
+ }
+
+ @AfterMethod(alwaysRun = true)
+ public void tearDown() {
+ try {
+ log.warn("Dropping test table {} ", table);
+ kustoAdminClient.execute(".drop table " + table + " ifexists");
+ } catch (Exception ignore) {
+ }
+ }
+
+ @Test
+ public void testOpenAndWriteSink() throws Exception {
+
+ ADXSink sink = new ADXSink();
+ sink.open(configs, Mockito.mock(SinkContext.class));
+ int writeCount = 50;
+
+ for (int i = 0; i < writeCount; i++) {
+ Record<byte[]> record = build("key_" + i, "test data from ADX Pulsar Sink_" + i);
+ sink.write(record);
+ }
+ Thread.sleep(40000);
+ KustoOperationResult result = kustoAdminClient.execute(database, table + " | count");
+ KustoResultSetTable mainTableResult = result.getPrimaryResults();
+ mainTableResult.next();
+ int actualRowsCount = mainTableResult.getInt(0);
+ Assert.assertEquals(actualRowsCount, writeCount);
+ kustoAdminClient.execute(database, ".clear table " + table + " data");
+ sink.close();
+ }
+
+ @Test
+ public void testOpenAndWriteSinkWithTimeouts() throws Exception {
+ ADXSink sink = new ADXSink();
+ sink.open(configs, Mockito.mock(SinkContext.class));
+ int writeCount = 9;
+
+ for (int i = 0; i < writeCount; i++) {
+ Record<byte[]> record = build("key_" + i, "test data from ADX Pulsar Sink_" + i);
+ sink.write(record);
+ }
+ Thread.sleep(40000);
+ KustoOperationResult result = kustoAdminClient.execute(database, table + " | count");
+ KustoResultSetTable mainTableResult = result.getPrimaryResults();
+ mainTableResult.next();
+ int actualRowsCount = mainTableResult.getInt(0);
+ Assert.assertEquals(actualRowsCount, writeCount);
+
+ sink.close();
+ }
+
+ private Record<byte[]> build(String key, String value) {
+ return new SinkRecord<byte[]>(new Record<>() {
+
+ @Override
+ public byte[] getValue() {
+ return value.getBytes(StandardCharsets.UTF_8);
+ }
+
+ @Override
+ public Optional<String> getDestinationTopic() {
+ return Optional.of("destination-topic");
+ }
+
+ @Override
+ public Optional<Long> getEventTime() {
+ return Optional.of(System.currentTimeMillis());
+ }
+
+ @Override
+ public Optional<String> getKey() {
+ return Optional.of("key-" + key);
+ }
+
+ @Override
+ public Map<String, String> getProperties() {
+ return new HashMap<>();
+ }
+ }, value.getBytes(StandardCharsets.UTF_8));
+ }
+}
diff --git a/pulsar-io/azure-data-explorer/src/test/resources/sinkConfig.yaml b/pulsar-io/azure-data-explorer/src/test/resources/sinkConfig.yaml
new file mode 100644
index 00000000000..32086a158f0
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/test/resources/sinkConfig.yaml
@@ -0,0 +1,33 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{
+ "clusterUrl":"https://somecluster.eastus.kusto.windows.net",
+ "database":"somedb",
+ "table": "tableName",
+ "appId": "xxxx-xxxx-xxxx-xxxx",
+ "appKey": "xxxx-xxxx-xxxx-xxxx",
+ "tenantId": "xxxx-xxxx-xxxx-xxxx",
+ "managedIdentityId": "xxxx-some-id-xxxx OR empty string",
+ "mappingRefName": "mapping ref name",
+ "mappingRefType":"CSV",
+ "flushImmediately":false,
+ "batchSize":100,
+ "batchTimeMs":10000,
+}
diff --git a/pulsar-io/azure-data-explorer/src/test/resources/sinkConfigInvalid.yaml b/pulsar-io/azure-data-explorer/src/test/resources/sinkConfigInvalid.yaml
new file mode 100644
index 00000000000..55537a90af8
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/test/resources/sinkConfigInvalid.yaml
@@ -0,0 +1,26 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{
+ "clusterUrl":"https://somecluster.eastus.kusto.windows.net",
+ "database":"somedb",
+ "table": "tableName",
+ "appId": "xxxx-xxxx-xxxx-xxxx",
+ "appKey": "xxxx-xxxx-xxxx-xxxx",
+}
diff --git a/pulsar-io/azure-data-explorer/src/test/resources/sinkConfigValid.yaml b/pulsar-io/azure-data-explorer/src/test/resources/sinkConfigValid.yaml
new file mode 100644
index 00000000000..32086a158f0
--- /dev/null
+++ b/pulsar-io/azure-data-explorer/src/test/resources/sinkConfigValid.yaml
@@ -0,0 +1,33 @@
+#
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+#
+
+{
+ "clusterUrl":"https://somecluster.eastus.kusto.windows.net",
+ "database":"somedb",
+ "table": "tableName",
+ "appId": "xxxx-xxxx-xxxx-xxxx",
+ "appKey": "xxxx-xxxx-xxxx-xxxx",
+ "tenantId": "xxxx-xxxx-xxxx-xxxx",
+ "managedIdentityId": "xxxx-some-id-xxxx OR empty string",
+ "mappingRefName": "mapping ref name",
+ "mappingRefType":"CSV",
+ "flushImmediately":false,
+ "batchSize":100,
+ "batchTimeMs":10000,
+}
diff --git a/pulsar-io/docs/pom.xml b/pulsar-io/docs/pom.xml
index 469056fb688..adbc2f4efad 100644
--- a/pulsar-io/docs/pom.xml
+++ b/pulsar-io/docs/pom.xml
@@ -57,6 +57,11 @@
<artifactId>pulsar-io-alluxio</artifactId>
<version>${project.version}</version>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-azuredataexplorer</artifactId>
+ <version>${project.version}</version>
+ </dependency>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-io-canal</artifactId>
diff --git a/pulsar-io/pom.xml b/pulsar-io/pom.xml
index 130baa3926d..b8d900dd060 100644
--- a/pulsar-io/pom.xml
+++ b/pulsar-io/pom.xml
@@ -22,7 +22,7 @@
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<packaging>pom</packaging>
- <parent>
+ <parent>
<groupId>org.apache.pulsar</groupId>
<artifactId>pulsar</artifactId>
<version>3.3.0-SNAPSHOT</version>
@@ -42,6 +42,7 @@
</property>
</activation>
<modules>
+ <module>azure-data-explorer</module>
<module>core</module>
<module>batch-discovery-triggerers</module>
<module>batch-data-generator</module>
@@ -81,6 +82,7 @@
<profile>
<id>pulsar-io-tests</id>
<modules>
+ <module>azure-data-explorer</module>
<module>core</module>
<module>batch-discovery-triggerers</module>
<module>batch-data-generator</module>