Azure Blob Storage Support 1311

[gkatzioura]

Purpose

Linked issue: close #1311

Add File System support for Azure Blob Storage.

Brief change log

Tests

API and Format

Documentation

[polyzos]

@gkatzioura, you mean -> Add File System support for Azure Blob Storage?

[gkatzioura]

@gkatzioura, you mean -> Add File System support for Azure Blob Storage?

Fixed

[luoyuxia]

Just as #993 , obtainSecurityToken method is required for client to access Azure Blob Storage when read from remote

[gkatzioura]

Just as #993 , obtainSecurityToken method is required for client to access Azure Blob Storage when read from remote

checking the options

[polyzos]

@gkatzioura, we should make sure it works for all prefixes, like
https://nightlies.apache.org/flink/flink-docs-master/docs/deployment/filesystems/azure/#azure-blob-storage

[polyzos]

abfs:// works fine during my tests.

I pushed one commit to disable tests, because we can't meet the test coverage here, and I'm figuring out the issue with the license.

[gkatzioura]

abfs:// works fine during my tests.

I pushed one commit to disable tests, because we can't meet the test coverage here, and I'm figuring out the issue with the license.

Yes forgot to mention that I added the support for all formats

[polyzos]

LicenseChecker fixed, and I just finished testing with all schemes. Everything works great 💪 🚀
@luoyuxia @michaelkoepf Let me know if you have any comments on this PR, before merging

[gkatzioura]

Hi @luoyuxia
some observations on the obtainSecurityToken functionality.

I do receive the tokens when I read fluss data as a client

try (LogScanner logScanner = table.newScan().createLogScanner()) {
            logScanner.subscribeFromBeginning(1);
            var record = logScanner.poll(Duration.ofSeconds(10));
            record.forEach(r -> _ );
}

Once log scanner is started the tokens start to be sent.
However I could not see the tokens being shared above fluss components (tablet-server coordinator server).
Similar to what happens with Flink JobManager tokens delegated to the TaskManager.
My assumption was that the token is shared among the fluss components then it is used with a
hadoop custom provider type (Depends on the hadoop driver).

[luoyuxia]

Hi @luoyuxia some observations on the obtainSecurityToken functionality.

I do receive the tokens when I read fluss data as a client

try (LogScanner logScanner = table.newScan().createLogScanner()) {
            logScanner.subscribeFromBeginning(1);
            var record = logScanner.poll(Duration.ofSeconds(10));
            record.forEach(r -> _ );
}

Once log scanner is started the tokens start to be sent. However I could not see the tokens being shared above fluss components (tablet-server coordinator server). Similar to what happens with Flink JobManager tokens delegated to the TaskManager. My assumption was that the token is shared among the fluss components then it is used with a hadoop custom provider type (Depends on the hadoop driver).

Hi, the security token is not shared above fluss components. Currently, it's only used by fluss-client, fluss-client require a token to read from remote directly.

[luoyuxia]

@gkatzioura Thanks for the greate work! Left minor comment

[polyzos]

@gkatzioura is there anything else pending here?

[gkatzioura]

@polyzos & @luoyuxia @wuchong fyi this is ready for the release

[leekeiabstraction]

Thank you for the PR! Hope you don't mind me reviewing, left a few comments. PTAL!

[wuchong]

@gkatzioura Thanks for the contribution! The implementation looks good to me. I only have some concerns about the naming conventions used in this PR:

1. Module/Artifact Name

The current fluss-fs-abfs uses a specific scheme name ("abfs") rather than the storage service name. Since this module supports multiple schemes (abfs, abfss, wasb, wasbs), the module should be named after the storage service: fluss-fs-azure. azure is more friendly than abfs for users when picking up the filesystem jars. Flink and Hadoop also use the same naming for the modules.

2. Package Name

Same issue applies to the package org.apache.fluss.fs.abfs. It should be: org.apache.fluss.fs.azure.

3. Class Names

It's weird that WasbFileSystemPlugin extends AbfsFileSystemPlugin. We should extract the common logic of AbfsFileSystemPlugin into an abstract class AzureFileSystemPlugin. And making specific azure scheme filesystem plugins extends the AzureFileSystemPlugin. Adding more detailed comment for the abstract class

/**
 * Abstract factory for creating Azure Blob Storage file systems. Supports multiple URI schemes
 * (abfs, abfss, wasb, wasbs) based on Azure HDFS support in the hadoop-azure module.
 */
abstract class AzureFileSystemPlugin implements FileSystemPlugin {

[wuchong]

This change introduces many Category B dependencies. According to Apache Software Foundation policy, all Category B dependencies require a corresponding LICENSE-xxx notice in the resource folder. These dependencies are likely pulled in because of a direct dependency on hadoop-common.

To avoid this, please try replacing the direct hadoop-common dependency with fluss-fs-hadoop-shaded. This shaded artifact should encapsulate those transitive dependencies and may eliminate the need to declare them explicitly in the NOTICE file.

[wuchong]

Also, since the branch is far behind main, could you please rebase it onto the latest main branch? This will make the CI results more reliable.