feat(filesystems): add support for Azure Blob Storage (#112)

This commit adds the new module filepulse-azure-blob-storage-fs that provides:
- AzureBlobStorageAvroFileInputReader
- AzureBlobStorageBytesArrayInputReader
- AzureBlobStorageRowFileInputReader
- AzureBlobStorageXMLFileInputReader
- AzureBlobStorageFileSystemListing

Delete ScELLexer.interp

refacto

Author: qgeffard

connect-file-pulse-api/src/main/java/io/streamthoughts/kafka/connect/filepulse/source/FileObjectMeta.java

@@ -29,7 +29,7 @@
 import java.util.Objects;
 
 /**
- * An object regrouping metadata about an input ile manipulate by the connector.
+ * An object regrouping metadata about an input file manipulate by the connector.
  */
 public interface FileObjectMeta extends Serializable, Comparable<FileObjectMeta> {
 

connect-file-pulse-filesystems/filepulse-amazons3-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/aws/AmazonS3ClientConfig.java

@@ -134,7 +134,7 @@ static ConfigDef getConf() {
                         GROUP_AWS,
                         awsGroupCounter++,
                         ConfigDef.Width.NONE,
-                        AWS_S3_BUCKET_NAME_CONFIG
+                        AWS_S3_BUCKET_PREFIX_CONFIG
                 )
 
                 .define(

connect-file-pulse-filesystems/filepulse-azure-blob-storage-fs/pom.xml

@@ -0,0 +1,47 @@
+<?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 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <groupId>io.streamthoughts</groupId>
+        <artifactId>kafka-connect-filepulse-filesystems</artifactId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+
+    <groupId>io.streamthoughts</groupId>
+    <artifactId>kafka-filepulse-azure-blob-storage-fs</artifactId>
+    <version>2.0.0-SNAPSHOT</version>
+    <description>Kafka Connect FilePulse - FileSystem - Support for Azure Blob Storage</description>
+
+    <properties>
+        <checkstyle.config.location>${project.parent.basedir}/..</checkstyle.config.location>
+        <azure-storage-blob.version>12.6.0</azure-storage-blob.version>
+    </properties>
+
+    <dependencies>
+        <dependency>
+            <groupId>com.azure</groupId>
+            <artifactId>azure-storage-blob</artifactId>
+            <version>${azure-storage-blob.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>io.streamthoughts</groupId>
+            <artifactId>kafka-connect-filepulse-commons-fs</artifactId>
+            <version>${project.version}</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>kafka-clients</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.kafka</groupId>
+            <artifactId>connect-api</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.avro</groupId>
+            <artifactId>avro</artifactId>
+        </dependency>
+    </dependencies>
+</project>

connect-file-pulse-filesystems/filepulse-azure-blob-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/azure/AzureBlobStorageClientConfig.java

@@ -0,0 +1,91 @@
+/*
+ * Copyright 2019-2021 StreamThoughts.
+ *
+ * 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 io.streamthoughts.kafka.connect.filepulse.storage.azure;
+
+import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigDef;
+import java.util.Map;
+
+
+public class AzureBlobStorageClientConfig extends AbstractConfig {
+
+    private static final String GROUP_AZURE = "AZURE";
+
+    public static final String AZURE_BLOB_STORAGE_CONNECTION_STRING_CONFIG = "azure.blob.storage.connection.string";
+    private static final String AZURE_BLOB_STORAGE_CONNECTION_STRING_DOC = "Azure storage account connection string";
+
+    public static final String AZURE_BLOB_STORAGE_CONTAINER_NAME_CONFIG = "azure.blob.storage.container.name";
+    private static final String AZURE_BLOB_STORAGE_CONTAINER_NAME_DOC = "The container name";
+
+    public final static String AZURE_BLOB_STORAGE_PREFIX_CONFIG = "azure.blob.storage.prefix";
+    private final static String AZURE_BLOB_STORAGE_PREFIX_DOC = "The prefix to be used for restricting the listing of the blobs in the container";
+
+    public AzureBlobStorageClientConfig(final Map<String, ?> originals) {
+        super(getConf(), originals, false);
+    }
+
+    /**
+     * @return the {@link ConfigDef}.
+     */
+    static ConfigDef getConf() {
+        int azureGroupCounter = 0;
+
+        return new ConfigDef()
+                .define(
+                        AZURE_BLOB_STORAGE_CONNECTION_STRING_CONFIG,
+                        ConfigDef.Type.STRING,
+                        ConfigDef.Importance.HIGH,
+                        AZURE_BLOB_STORAGE_CONNECTION_STRING_DOC,
+                        GROUP_AZURE,
+                        azureGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        AZURE_BLOB_STORAGE_CONNECTION_STRING_CONFIG
+                )
+                .define(
+                        AZURE_BLOB_STORAGE_CONTAINER_NAME_CONFIG,
+                        ConfigDef.Type.STRING,
+                        ConfigDef.Importance.HIGH,
+                        AZURE_BLOB_STORAGE_CONTAINER_NAME_DOC,
+                        GROUP_AZURE,
+                        azureGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        AZURE_BLOB_STORAGE_CONTAINER_NAME_CONFIG
+                )
+                .define(
+                        AZURE_BLOB_STORAGE_PREFIX_CONFIG,
+                        ConfigDef.Type.STRING,
+                        ConfigDef.Importance.HIGH,
+                        AZURE_BLOB_STORAGE_PREFIX_DOC,
+                        GROUP_AZURE,
+                        azureGroupCounter++,
+                        ConfigDef.Width.NONE,
+                        AZURE_BLOB_STORAGE_PREFIX_CONFIG
+                );
+    }
+
+    public String getConnectionString() {
+        return getString(AZURE_BLOB_STORAGE_CONNECTION_STRING_CONFIG);
+    }
+
+    public String getContainerName() {
+        return getString(AZURE_BLOB_STORAGE_CONTAINER_NAME_CONFIG);
+    }
+
+    public String getAzureBlobStoragePrefix() {return getString(AZURE_BLOB_STORAGE_PREFIX_CONFIG);}
+}

connect-file-pulse-filesystems/filepulse-azure-blob-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/azure/AzureBlobStorageClientUtils.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2019-2021 StreamThoughts.
+ *
+ * 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 io.streamthoughts.kafka.connect.filepulse.storage.azure;
+
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.BlobServiceClientBuilder;
+
+/**
+ * Utility class for creating new {@link BlobContainerClient} client.
+ */
+public class AzureBlobStorageClientUtils {
+
+    public static BlobContainerClient createBlobContainerClient(AzureBlobStorageClientConfig config){
+        // Create a BlobServiceClient object which is used to create a container client
+        return new BlobServiceClientBuilder()
+                .connectionString(config.getConnectionString()).buildClient()
+                .getBlobContainerClient(config.getContainerName());
+    }
+}

connect-file-pulse-filesystems/filepulse-azure-blob-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/azure/blob/AzureBlobStorage.java

@@ -0,0 +1,89 @@
+/*
+ * Copyright 2019-2021 StreamThoughts.
+ *
+ * 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 io.streamthoughts.kafka.connect.filepulse.storage.azure.blob;
+
+import com.azure.storage.blob.BlobClient;
+import com.azure.storage.blob.BlobClientBuilder;
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.models.BlobItem;
+import io.streamthoughts.kafka.connect.filepulse.fs.reader.Storage;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
+import io.streamthoughts.kafka.connect.filepulse.source.GenericFileObjectMeta;
+
+import java.io.InputStream;
+import java.net.URI;
+import java.util.Arrays;
+import java.util.HashMap;
+
+public class AzureBlobStorage implements Storage {
+
+    private final BlobContainerClient containerClient;
+
+    public AzureBlobStorage(BlobContainerClient blobContainerClient) {
+        this.containerClient = blobContainerClient;
+    }
+
+    @Override
+    public FileObjectMeta getObjectMetadata(URI uri) {
+        BlobClient blobClient = getBlobClient(uri);
+
+        return new GenericFileObjectMeta.Builder()
+                .withUri(uri)
+                .withName(blobClient.getBlobName())
+                .withContentLength(blobClient.getProperties().getBlobSize())
+                .withLastModified(blobClient.getProperties().getLastModified().toInstant())
+                .withContentDigest(
+                        new FileObjectMeta.ContentDigest(
+                                Arrays.toString(blobClient.getProperties().getContentMd5()), "MD5"))
+                .withUserDefinedMetadata(new HashMap<>(blobClient.getProperties().getMetadata()))
+                .build();
+    }
+
+    public FileObjectMeta getObjectMetadata(BlobItem blobItem, URI uri) {
+        return new GenericFileObjectMeta.Builder()
+                .withUri(uri)
+                .withName(blobItem.getName())
+                .withContentLength(blobItem.getProperties().getContentLength())
+                .withLastModified(blobItem.getProperties().getLastModified().toInstant())
+                .withContentDigest(
+                        new FileObjectMeta.ContentDigest(
+                                Arrays.toString(blobItem.getProperties().getContentMd5()), "MD5"))
+                .withUserDefinedMetadata(new HashMap<>(blobItem.getMetadata()))
+                .build();
+    }
+
+    @Override
+    public boolean exists(URI uri) {
+        return getBlobClient(uri).exists();
+    }
+
+    @Override
+    public InputStream getInputStream(URI uri) {
+        return getBlobClient(uri).openInputStream();
+    }
+
+    private BlobClient getBlobClient(URI uri) {
+        return containerClient
+                .getBlobClient(new BlobClientBuilder().endpoint(uri.toString()).buildClient().getBlobName());
+    }
+
+    public BlobContainerClient getBlobContainerClient() {
+        return containerClient;
+    }
+}

connect-file-pulse-filesystems/filepulse-azure-blob-storage-fs/src/main/java/io/streamthoughts/kafka/connect/filepulse/storage/azure/blob/AzureBlobStorageFileSystemListing.java

@@ -0,0 +1,86 @@
+/*
+ * Copyright 2019-2021 StreamThoughts.
+ *
+ * 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 io.streamthoughts.kafka.connect.filepulse.storage.azure.blob;
+
+import com.azure.storage.blob.BlobContainerClient;
+import com.azure.storage.blob.models.BlobListDetails;
+import com.azure.storage.blob.models.ListBlobsOptions;
+import io.streamthoughts.kafka.connect.filepulse.fs.FileListFilter;
+import io.streamthoughts.kafka.connect.filepulse.fs.FileSystemListing;
+import io.streamthoughts.kafka.connect.filepulse.source.FileObjectMeta;
+import io.streamthoughts.kafka.connect.filepulse.storage.azure.AzureBlobStorageClientConfig;
+import io.streamthoughts.kafka.connect.filepulse.storage.azure.AzureBlobStorageClientUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.net.URI;
+import java.net.URISyntaxException;
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+import static io.streamthoughts.kafka.connect.filepulse.internal.StringUtils.isNotBlank;
+
+public class AzureBlobStorageFileSystemListing implements FileSystemListing {
+
+    private static final Logger LOG = LoggerFactory.getLogger(AzureBlobStorageFileSystemListing.class);
+
+    private FileListFilter filter;
+    private AzureBlobStorage storage;
+    private AzureBlobStorageClientConfig config;
+
+    @Override
+    public void configure(final Map<String, ?> configs) {
+        this.config = new AzureBlobStorageClientConfig(configs);
+        storage = new AzureBlobStorage(
+                AzureBlobStorageClientUtils.createBlobContainerClient(config));
+    }
+
+    @Override
+    public Collection<FileObjectMeta> listObjects() {
+        BlobContainerClient blobContainerClient = storage.getBlobContainerClient();
+        ListBlobsOptions options = new ListBlobsOptions();
+        if (isNotBlank(config.getAzureBlobStoragePrefix()))
+            options.setPrefix(config.getAzureBlobStoragePrefix());
+        List<FileObjectMeta> fileObjectMetaList = blobContainerClient.listBlobs(options, null)
+                .stream()
+                .map(blobItem -> {
+                    // URI construction, as the azure sdk sucks...
+                    String blobUrl = blobContainerClient.getBlobClient(blobItem.getName()).getBlobUrl();
+                    URI uri = null;
+                    try {
+                        uri = new URI(blobUrl);
+                    } catch (URISyntaxException e) {
+                        LOG.error(
+                                "Failed to construct the blob URI from the given URL : '{}'. ",
+                                blobUrl,
+                                e
+                        );
+                    }
+                    return storage.getObjectMetadata(blobItem, uri);
+                }).collect(Collectors.toList());
+        return filter == null ? fileObjectMetaList : filter.filterFiles(fileObjectMetaList);
+    }
+
+    @Override
+    public void setFilter(FileListFilter filter) {
+        this.filter = filter;
+    }
+}