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