[Bug]: python flink runner is not compatible with Azure blob file system in Java
- Dominant language
- Java
- Stars
- 8.7k
- Forks
- 4.7k
- Avg merge
- 1d 20h
- Merged PRs (30d)
- 196
Description
### What happened?
1. Python [flink_runner starts](https://github.com/apache/beam/blob/master/sdks/python/apache_beam/runners/portability/flink_runner.py#L109) beam flink job-server with these parameters:
```python
def java_arguments(
self, job_port, artifact_port, expansion_port, artifacts_dir):
return [
'--flink-master',
self._master_url,
'--artifacts-dir',
artifacts_dir,
'--job-port',
job_port,
'--artifact-port',
artifact_port,
'--expansion-port',
expansion_port
]
```
2. When **artifacts_dir** above is path to a container in Azure blob storage (e.g. "azfs://storage-account/container), the [AzureBlobStoreFileSystem](https://github.com/apache/beam/blob/4ffeae4d2b800f2df36d2ea2eab549f2204d5691/sdks/java/io/azure/src/main/java/org/apache/beam/sdk/io/azure/blobstore/AzureBlobStoreFileSystem.java#L271) attempts to create BlobServiceClient and fails with:
```
I1206 02:34:29.021346 139656344459008 subprocess_server.py:126] Caused by: java.lang.IllegalArgumentException: Invalid URL format. URL: null
I1206 02:34:29.021379 139656344459008 subprocess_server.py:126] at com.azure.storage.blob.BlobUrlParts.parse(BlobUrlParts.java:349)
I1206 02:34:29.021412 139656344459008 subprocess_server.py:126] at com.azure.storage.blob.implementation.util.BuilderHelper.httpsValidation(BuilderHelper.java:170)
I1206 02:34:29.021589 139656344459008 subprocess_server.py:126] at com.azure.storage.blob.implementation.util.BuilderHelper.buildPipeline(BuilderHelper.java:106)
I1206 02:34:29.024176 139656344459008 subprocess_server.py:126] at com.azure.storage.blob.BlobServiceClientBuilder.buildAsyncClient(BlobServiceClientBuilder.java:113)
I1206 02:34:29.024224 139656344459008 subprocess_server.py:126] at com.azure.storage.blob.BlobServiceClientBuilder.buildClient(BlobServiceClientBuilder.java:89)
I1206 02:34:29.024258 139656344459008 subprocess_server.py:126] at org.apache.beam.vendor.guava.v26_0_jre.com.google.common.base.Suppliers$NonSerializableMemoizingSupplier.get(Suppliers.java:167)
I1206 02:34:29.024288 139656344459008 subprocess_server.py:126] at org.apache.beam.sdk.io.azure.blobstore.AzureBlobStoreFileSystem.create(AzureBlobStoreFileSystem.java:271)
I1206 02:34:29.024316 139656344459008 subprocess_server.py:126] at org.apache.beam.sdk.io.azure.blobstore.AzureBlobStoreFileSystem.create(AzureBlobStoreFileSystem.java:68)
I1206 02:34:29.024344 139656344459008 subprocess_server.py:126] at org.apache.beam.sdk.io.FileSystems.create(FileSystems.java:243)
I1206 02:34:29.024372 139656344459008 subprocess_server.py:126] at org.apache.beam.sdk.io.FileSystems.create(FileSystems.java:230)
I1206 02:34:29.024399 139656344459008 subprocess_server.py:126] at org.apache.beam.runners.fnexecution.artifact.ArtifactStagingService$ArtifactDestination.fromFile(ArtifactStagingService.java:140)
```
The null reference exception occurs because the endpoint to blob service is not configured. It is expected to be passed via "blob_service_endpoint" [option](https://github.com/apache/beam/blob/243128a8fc52798e1b58b0cf1a271d95ee7aa241/sdks/java/io/azure/src/main/java/org/apache/beam/sdk/io/azure/options/BlobstoreOptions.java#L85). Neither the option is passed from flink_runner.py nor it is inferred from the "azfs://storage-account/container-name" file path in AzureBlobStoreFileSystem.java.
### Issue Priority
Priority: 1
### Issue Component
Component: beam-community
Contributor guide
Assessment
This issue has not been assessed yet.