Add serverless provisioner

[sahusanket]

No description provided.

[gemini-code-assist]

Code Review

This pull request introduces support for Dataproc Serverless by adding the DataprocServerlessProvisioner, DataprocServerlessRuntimeJobManager, configuration metadata, and corresponding unit tests. Feedback on the changes highlights three critical issues in DataprocServerlessRuntimeJobManager: a potential NullPointerException in validateDeleteLifecycle if the GCS bucket does not exist, a bypass of the configured Storage client in openStream, and a temporary directory leak in /tmp when local caching is disabled.

[gemini-code-assist]

If the GCS bucket does not exist or is inaccessible, storage.get(bucketName) will return null. Accessing bucket.getLifecycleRules() without a null check will throw a NullPointerException. A null check should be added to handle this defensively.

  private boolean validateDeleteLifecycle(String bucketName, String run) {
    Storage storage = getStorageClient();
    Bucket bucket = storage.get(bucketName);
    if (bucket == null) {
      return false;
    }
    for (BucketInfo.LifecycleRule rule : bucket.getLifecycleRules()) {

[gemini-code-assist]

Using StorageOptions.getDefaultInstance().getService() bypasses the configured Storage client (which has custom credentials, project ID, timeouts, and retry settings). This can lead to authentication/authorization failures or connection timeouts in environments where default credentials are not configured or differ from the provisioner's credentials. Use getStorageClient() instead.

Suggested change

	if ("gs".equals(uri.getScheme())) {
	BlobId blobId = BlobId.of(uri.getAuthority(), uri.getPath().substring(1));
	Storage client = StorageOptions.getDefaultInstance().getService();
	return Channels.newInputStream(client.get(blobId).reader());
	}
	if ("gs".equals(uri.getScheme())) {
	BlobId blobId = BlobId.of(uri.getAuthority(), uri.getPath().substring(1));
	Storage client = getStorageClient();
	return Channels.newInputStream(client.get(blobId).reader());
	}

[gemini-code-assist]

When disableLocalCaching is enabled, a temporary directory is created using Files.createTempDirectory. However, DataprocUtils.deleteDirectoryContents(tempDir) only deletes the files and subdirectories inside it, leaving the empty temporary directory itself behind. This causes a directory leak in /tmp. Call tempDir.delete() after deleting its contents to clean up the directory itself.

      if (disableLocalCaching) {
        DataprocUtils.deleteDirectoryContents(tempDir);
        tempDir.delete();
      }