[wip]feat: Spark Serverless #1036
Open
Conversation
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
WalkthroughSupport for a new job type, Changes
Sequence Diagram(s)sequenceDiagram
participant User
participant DataprocSubmitter
participant BatchControllerClient
User->>DataprocSubmitter: submit(jobType=SparkServerlessJob, ...)
DataprocSubmitter->>DataprocSubmitter: buildSparkServerlessJob(...)
DataprocSubmitter->>BatchControllerClient: submitBatchAsync(batch)
BatchControllerClient-->>DataprocSubmitter: batchName
DataprocSubmitter-->>User: batchName
Estimated code review effort🎯 3 (Moderate) | ⏱️ ~15 minutes Possibly related PRs
Suggested reviewers
Poem
✨ Finishing Touches🧪 Generate unit tests
🪧 TipsChatThere are 3 ways to chat with CodeRabbit:
SupportNeed help? Create a ticket on our support page for assistance with any issues or questions. CodeRabbit Commands (Invoked using PR comments)
Other keywords and placeholders
CodeRabbit Configuration File (
|
![coderabbitai[bot]](https://avatars.githubusercontent.com/in/347564?s=60&v=4){kind=small}
There was a problem hiding this comment.
Actionable comments posted: 3
🧹 Nitpick comments (1)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala (1)
403-404: Reduce code duplicationSparkServerlessJob cases duplicate SparkJob logic exactly.
val submissionProps = jobType match { - case TypeSparkServerlessJob => - Map(MainClass -> mainClass, JarURI -> jarUri, MetadataName -> metadataName, JobId -> jobId) - case TypeSparkJob => + case TypeSparkServerlessJob | TypeSparkJob => Map(MainClass -> mainClass, JarURI -> jarUri, MetadataName -> metadataName, JobId -> jobId)Similarly for getApplicationArgs:
val finalArgs = jobType match { - case TypeSparkJob => { - val bigtableInstanceId = envMap(GcpBigtableInstanceIdEnvVar).getOrElse("") - val projectId = envMap(GcpProjectIdEnvVar).getOrElse(throw new Exception(s"GcpProjectId not set")) - Array.concat(userArgs) - } + case TypeSparkJob | TypeSparkServerlessJob => { + val bigtableInstanceId = envMap(GcpBigtableInstanceIdEnvVar).getOrElse("") + val projectId = envMap(GcpProjectIdEnvVar).getOrElse(throw new Exception(s"GcpProjectId not set")) + Array.concat(userArgs) + } case TypeFlinkJob => { val additionalArgsToFilterOut = Set( ConfTypeArgKeyword ) userArgs.filter(arg => !additionalArgsToFilterOut.exists(arg.startsWith)) } - case TypeSparkServerlessJob => { - val bigtableInstanceId = envMap(GcpBigtableInstanceIdEnvVar).getOrElse("") - val projectId = envMap(GcpProjectIdEnvVar).getOrElse(throw new Exception(s"GcpProjectId not set")) - Array.concat(userArgs) - } }Also applies to: 494-499
📜 Review details
Configuration used: CodeRabbit UI
Review profile: CHILL
Plan: Pro
📒 Files selected for processing (3)
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala(5 hunks)cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala(1 hunks)spark/src/main/scala/ai/chronon/spark/submission/JobSubmitter.scala(2 hunks)
🧰 Additional context used
🧠 Learnings (8)
📚 Learning: the import `scala.util.scalaversionspecificcollectionsconverter` in `service/src/main/java/ai/chrono...
Learnt from: nikhil-zlai
PR: zipline-ai/chronon#70
File: service/src/main/java/ai/chronon/service/ApiProvider.java:6-6
Timestamp: 2024-12-03T04:04:33.809Z
Learning: The import `scala.util.ScalaVersionSpecificCollectionsConverter` in `service/src/main/java/ai/chronon/service/ApiProvider.java` is correct and should not be flagged in future reviews.
Applied to files:
spark/src/main/scala/ai/chronon/spark/submission/JobSubmitter.scalacloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scalacloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala
📚 Learning: in aws emr, the term "job" is ambiguous and can refer to either a step (single spark execution) or a...
Learnt from: david-zlai
PR: zipline-ai/chronon#439
File: cloud_aws/src/main/scala/ai/chronon/integrations/aws/EmrSubmitter.scala:198-206
Timestamp: 2025-03-07T20:41:11.525Z
Learning: In AWS EMR, the term "job" is ambiguous and can refer to either a Step (single Spark execution) or a JobFlow/Cluster. EMR operations typically require both a clusterId and a stepId, while the JobSubmitter interface expects a single jobId parameter.
Applied to files:
spark/src/main/scala/ai/chronon/spark/submission/JobSubmitter.scalacloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala
📚 Learning: in `mockkvstore` located at `spark/src/test/scala/ai/chronon/spark/test/mockkvstore.scala`, the `mul...
Learnt from: chewy-zlai
PR: zipline-ai/chronon#50
File: spark/src/test/scala/ai/chronon/spark/test/MockKVStore.scala:19-28
Timestamp: 2024-10-31T18:29:45.027Z
Learning: In `MockKVStore` located at `spark/src/test/scala/ai/chronon/spark/test/MockKVStore.scala`, the `multiPut` method is intended to be a simple implementation without dataset existence validation, duplicate validation logic elimination, or actual storage of key-value pairs for verification.
Applied to files:
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala
📚 Learning: references to `mockdataservice` in `hub/test/controllers/searchcontrollerspec.scala` and `hub/test/c...
Learnt from: piyush-zlai
PR: zipline-ai/chronon#44
File: hub/app/controllers/ModelController.scala:15-18
Timestamp: 2024-10-17T19:46:42.629Z
Learning: References to `MockDataService` in `hub/test/controllers/SearchControllerSpec.scala` and `hub/test/controllers/ModelControllerSpec.scala` are needed for tests and should not be removed.
Applied to files:
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala
📚 Learning: hardcoding future timestamps in tests within `dynamodbkvstoretest.scala` is acceptable when data is ...
Learnt from: piyush-zlai
PR: zipline-ai/chronon#33
File: cloud_aws/src/test/scala/ai/chronon/integrations/aws/DynamoDBKVStoreTest.scala:175-175
Timestamp: 2024-10-07T15:09:51.567Z
Learning: Hardcoding future timestamps in tests within `DynamoDBKVStoreTest.scala` is acceptable when data is generated and queried within the same time range, ensuring the tests remain valid over time.
Applied to files:
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala
📚 Learning: in `hub/test/store/dynamodbmonitoringstoretest.scala`, the current implementation of the `generateli...
Learnt from: piyush-zlai
PR: zipline-ai/chronon#44
File: hub/test/store/DynamoDBMonitoringStoreTest.scala:69-86
Timestamp: 2024-10-15T15:33:22.265Z
Learning: In `hub/test/store/DynamoDBMonitoringStoreTest.scala`, the current implementation of the `generateListResponse` method is acceptable as-is, and changes for resource handling and error management are not necessary at this time.
Applied to files:
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala
📚 Learning: in spark applications, when defining serializable classes, passing an implicit `executioncontext` pa...
Learnt from: chewy-zlai
PR: zipline-ai/chronon#62
File: spark/src/main/scala/ai/chronon/spark/stats/drift/SummaryUploader.scala:9-10
Timestamp: 2024-11-06T21:54:56.160Z
Learning: In Spark applications, when defining serializable classes, passing an implicit `ExecutionContext` parameter can cause serialization issues. In such cases, it's acceptable to use `scala.concurrent.ExecutionContext.Implicits.global`.
Applied to files:
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala
📚 Learning: for bigquery table creation operations in bigqueryformat.scala, allow exceptions to propagate direct...
Learnt from: tchow-zlai
PR: zipline-ai/chronon#263
File: cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/BigQueryFormat.scala:56-57
Timestamp: 2025-01-24T23:55:40.650Z
Learning: For BigQuery table creation operations in BigQueryFormat.scala, allow exceptions to propagate directly without wrapping them in try-catch blocks, as the original BigQuery exceptions provide sufficient context.
Applied to files:
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala
⏰ Context from checks skipped due to timeout of 90000ms. You can increase the timeout in your CodeRabbit configuration to a maximum of 15 minutes (900000ms). (18)
- GitHub Check: scala_compile_fmt_fix
- GitHub Check: groupby_tests
- GitHub Check: analyzer_tests
- GitHub Check: streaming_tests
- GitHub Check: cloud_aws_tests
- GitHub Check: cloud_gcp_tests
- GitHub Check: batch_tests
- GitHub Check: fetcher_tests
- GitHub Check: service_tests
- GitHub Check: api_tests
- GitHub Check: service_commons_tests
- GitHub Check: online_tests
- GitHub Check: spark_tests
- GitHub Check: join_tests
- GitHub Check: python_tests
- GitHub Check: aggregator_tests
- GitHub Check: flink_tests
- GitHub Check: enforce_triggered_workflows
🔇 Additional comments (1)
spark/src/main/scala/ai/chronon/spark/submission/JobSubmitter.scala (1)
16-16: LGTM!New job type follows existing patterns.
Also applies to: 157-157
Comment on lines
+163
to
+170
| if (jobType.equals(TypeSparkServerlessJob)) { | ||
| val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, args: _*) | ||
| batch.getSparkBatch.getArgsList.asScala | ||
| val batchClient = BatchControllerClient.create() | ||
| val locationName = LocationName.of(conf.projectId, conf.region) | ||
| val batchF = batchClient.createBatchAsync(locationName, batch, jobId) | ||
| batchF.get.getName | ||
| } else { |
There was a problem hiding this comment.
Resource leak: BatchControllerClient not closed
The BatchControllerClient should be properly closed after use.
if (jobType.equals(TypeSparkServerlessJob)) {
val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, args: _*)
batch.getSparkBatch.getArgsList.asScala
- val batchClient = BatchControllerClient.create()
- val locationName = LocationName.of(conf.projectId, conf.region)
- val batchF = batchClient.createBatchAsync(locationName, batch, jobId)
- batchF.get.getName
+ val batchClient = BatchControllerClient.create()
+ try {
+ val locationName = LocationName.of(conf.projectId, conf.region)
+ val batchF = batchClient.createBatchAsync(locationName, batch, jobId)
+ batchF.get.getName
+ } finally {
+ batchClient.close()
+ }📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| if (jobType.equals(TypeSparkServerlessJob)) { | |
| val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, args: _*) | |
| batch.getSparkBatch.getArgsList.asScala | |
| val batchClient = BatchControllerClient.create() | |
| val locationName = LocationName.of(conf.projectId, conf.region) | |
| val batchF = batchClient.createBatchAsync(locationName, batch, jobId) | |
| batchF.get.getName | |
| } else { | |
| if (jobType.equals(TypeSparkServerlessJob)) { | |
| val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, args: _*) | |
| batch.getSparkBatch.getArgsList.asScala | |
| val batchClient = BatchControllerClient.create() | |
| try { | |
| val locationName = LocationName.of(conf.projectId, conf.region) | |
| val batchF = batchClient.createBatchAsync(locationName, batch, jobId) | |
| batchF.get.getName | |
| } finally { | |
| batchClient.close() | |
| } | |
| } else { |
🤖 Prompt for AI Agents
In
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala
around lines 163 to 170, the BatchControllerClient instance is created but not
closed, causing a resource leak. Fix this by ensuring the BatchControllerClient
is closed after use, preferably by using a try-finally block or a resource
management construct to close the client once the batch creation is complete.
Comment on lines
+237
to
+261
| private def buildSparkServerlessJob(mainClass: String, | ||
| jarUri: String, | ||
| files: List[String], | ||
| jobProperties: Map[String, String], | ||
| args: String*): Batch = { | ||
| val sparkJob = SparkBatch | ||
| .newBuilder() | ||
| .setMainClass(mainClass) | ||
| .setMainJarFileUri(jarUri) | ||
| .addJarFileUris(jarUri) | ||
| .addAllFileUris(files.asJava) | ||
| .addAllArgs(args.toIterable.asJava) | ||
| .build() | ||
| val runtimeConf = RuntimeConfig | ||
| .newBuilder() | ||
| .setVersion("2.3") | ||
| .putAllProperties(jobProperties.asJava) | ||
| Batch | ||
| .newBuilder() | ||
| .setName("batch-dev") | ||
| .setUuid(java.util.UUID.randomUUID().toString) | ||
| .setSparkBatch(sparkJob) | ||
| .setRuntimeConfig(runtimeConf) | ||
| .build() | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Fix hardcoded values and UUID handling
Several issues in buildSparkServerlessJob:
- Hardcoded batch name "batch-dev"
- Generates new UUID instead of using provided jobId
- Hardcoded runtime version "2.3"
private def buildSparkServerlessJob(mainClass: String,
jarUri: String,
files: List[String],
jobProperties: Map[String, String],
+ jobId: String,
args: String*): Batch = {
val sparkJob = SparkBatch
.newBuilder()
.setMainClass(mainClass)
.setMainJarFileUri(jarUri)
.addJarFileUris(jarUri)
.addAllFileUris(files.asJava)
.addAllArgs(args.toIterable.asJava)
.build()
val runtimeConf = RuntimeConfig
.newBuilder()
- .setVersion("2.3")
+ .setVersion(jobProperties.getOrElse("spark.runtime.version", "2.3"))
.putAllProperties(jobProperties.asJava)
Batch
.newBuilder()
- .setName("batch-dev")
- .setUuid(java.util.UUID.randomUUID().toString)
+ .setName(s"batch-$jobId")
+ .setUuid(jobId)
.setSparkBatch(sparkJob)
.setRuntimeConfig(runtimeConf)
.build()
}Also update the call site:
-val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, args: _*)
+val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, jobId, args: _*)📝 Committable suggestion
‼️ IMPORTANT
Carefully review the code before committing. Ensure that it accurately replaces the highlighted code, contains no missing lines, and has no issues with indentation. Thoroughly test & benchmark the code to ensure it meets the requirements.
Suggested change
| private def buildSparkServerlessJob(mainClass: String, | |
| jarUri: String, | |
| files: List[String], | |
| jobProperties: Map[String, String], | |
| args: String*): Batch = { | |
| val sparkJob = SparkBatch | |
| .newBuilder() | |
| .setMainClass(mainClass) | |
| .setMainJarFileUri(jarUri) | |
| .addJarFileUris(jarUri) | |
| .addAllFileUris(files.asJava) | |
| .addAllArgs(args.toIterable.asJava) | |
| .build() | |
| val runtimeConf = RuntimeConfig | |
| .newBuilder() | |
| .setVersion("2.3") | |
| .putAllProperties(jobProperties.asJava) | |
| Batch | |
| .newBuilder() | |
| .setName("batch-dev") | |
| .setUuid(java.util.UUID.randomUUID().toString) | |
| .setSparkBatch(sparkJob) | |
| .setRuntimeConfig(runtimeConf) | |
| .build() | |
| } | |
| private def buildSparkServerlessJob(mainClass: String, | |
| jarUri: String, | |
| files: List[String], | |
| jobProperties: Map[String, String], | |
| jobId: String, | |
| args: String*): Batch = { | |
| val sparkJob = SparkBatch | |
| .newBuilder() | |
| .setMainClass(mainClass) | |
| .setMainJarFileUri(jarUri) | |
| .addJarFileUris(jarUri) | |
| .addAllFileUris(files.asJava) | |
| .addAllArgs(args.toIterable.asJava) | |
| .build() | |
| val runtimeConf = RuntimeConfig | |
| .newBuilder() | |
| .setVersion(jobProperties.getOrElse("spark.runtime.version", "2.3")) | |
| .putAllProperties(jobProperties.asJava) | |
| Batch | |
| .newBuilder() | |
| .setName(s"batch-$jobId") | |
| .setUuid(jobId) | |
| .setSparkBatch(sparkJob) | |
| .setRuntimeConfig(runtimeConf) | |
| .build() | |
| } |
Suggested change
| private def buildSparkServerlessJob(mainClass: String, | |
| jarUri: String, | |
| files: List[String], | |
| jobProperties: Map[String, String], | |
| args: String*): Batch = { | |
| val sparkJob = SparkBatch | |
| .newBuilder() | |
| .setMainClass(mainClass) | |
| .setMainJarFileUri(jarUri) | |
| .addJarFileUris(jarUri) | |
| .addAllFileUris(files.asJava) | |
| .addAllArgs(args.toIterable.asJava) | |
| .build() | |
| val runtimeConf = RuntimeConfig | |
| .newBuilder() | |
| .setVersion("2.3") | |
| .putAllProperties(jobProperties.asJava) | |
| Batch | |
| .newBuilder() | |
| .setName("batch-dev") | |
| .setUuid(java.util.UUID.randomUUID().toString) | |
| .setSparkBatch(sparkJob) | |
| .setRuntimeConfig(runtimeConf) | |
| .build() | |
| } | |
| - val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, args: _*) | |
| + val batch = buildSparkServerlessJob(mainClass, jarUri, files, jobProperties, jobId, args: _*) |
🤖 Prompt for AI Agents
In
cloud_gcp/src/main/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitter.scala
lines 237 to 261, the buildSparkServerlessJob method uses hardcoded values for
batch name ("batch-dev") and runtime version ("2.3"), and generates a new UUID
instead of using a provided jobId. Modify the method to accept jobId, batchName,
and runtimeVersion as parameters, replace the hardcoded values with these
parameters, and use the provided jobId for the UUID field. Also update all call
sites of this method to pass the appropriate arguments accordingly.
Comment on lines
+1120
to
+1146
| it should "test spark serverless" in { | ||
| val submitterConf = SubmitterConf("canary-443022", "us-central1", "zipline-canary-cluster") | ||
|
|
||
| val baseArgs = Seq( | ||
| s"$JarUriArgKeyword=gs://zipline-artifacts-dev/release/latest/jars/cloud_gcp_lib_deploy.jar", | ||
| s"$JobTypeArgKeyword=spark-serverless", | ||
| s"$MainClassKeyword=ai.chronon.spark.batch.BatchNodeRunner", | ||
| s"$ZiplineVersionArgKeyword=latest", | ||
| s"--job-id=123", | ||
| s"$FilesArgKeyword=gs://zipline-warehouse-dev/metadata/execution/b473f979-5ce0-4046-beea-b9e991933d13/gcp.training_set.v1_test__0__backfill", | ||
| "--conf-path=gcp.training_set.v1_test__0__backfill", | ||
| "--start-ds=2023-11-15", | ||
| "--end-ds=2023-11-15", | ||
| "--online-class=ai.chronon.integrations.cloud_gcp.GcpApiImpl", | ||
| "--table-partitions-dataset=TABLE_PARTITIONS_thomaschow", | ||
| "-ZGCP_PROJECT_ID=canary-443022", | ||
| "-ZGCP_BIGTABLE_INSTANCE_ID=zipline-canary-instance", | ||
| "-ZENABLE_UPLOAD_CLIENTS=true" | ||
| ) | ||
| val submitter = DataprocSubmitter(submitterConf) | ||
| DataprocSubmitter.run( | ||
| args = baseArgs.toArray, | ||
| submitter = submitter, | ||
| envMap = Map("GCP_BIGTABLE_INSTANCE_ID" -> Option("zipline-canary-instance"), | ||
| "GCP_PROJECT_ID" -> Option("canary-443022")) | ||
| ) | ||
| } |
There was a problem hiding this comment.
🛠️ Refactor suggestion
Test needs assertions and mocking
The test lacks assertions and uses hardcoded GCP resources. Consider:
- Adding assertions to verify expected behavior
- Mocking DataprocSubmitter to avoid real GCP calls
- Using more descriptive test name like "should submit spark serverless job successfully"
- it should "test spark serverless" in {
- val submitterConf = SubmitterConf("canary-443022", "us-central1", "zipline-canary-cluster")
+ it should "submit spark serverless job successfully" in {
+ val submitterConf = SubmitterConf("test-project", "test-region", "test-cluster")
val baseArgs = Seq(
s"$JarUriArgKeyword=gs://zipline-artifacts-dev/release/latest/jars/cloud_gcp_lib_deploy.jar",
s"$JobTypeArgKeyword=spark-serverless",
s"$MainClassKeyword=ai.chronon.spark.batch.BatchNodeRunner",
s"$ZiplineVersionArgKeyword=latest",
s"--job-id=123",
s"$FilesArgKeyword=gs://zipline-warehouse-dev/metadata/execution/b473f979-5ce0-4046-beea-b9e991933d13/gcp.training_set.v1_test__0__backfill",
"--conf-path=gcp.training_set.v1_test__0__backfill",
"--start-ds=2023-11-15",
"--end-ds=2023-11-15",
"--online-class=ai.chronon.integrations.cloud_gcp.GcpApiImpl",
"--table-partitions-dataset=TABLE_PARTITIONS_thomaschow",
- "-ZGCP_PROJECT_ID=canary-443022",
- "-ZGCP_BIGTABLE_INSTANCE_ID=zipline-canary-instance",
+ "-ZGCP_PROJECT_ID=test-project",
+ "-ZGCP_BIGTABLE_INSTANCE_ID=test-instance",
"-ZENABLE_UPLOAD_CLIENTS=true"
)
- val submitter = DataprocSubmitter(submitterConf)
+ val submitter = mock[DataprocSubmitter]
+ when(submitter.getConf).thenReturn(submitterConf)
+ when(submitter.submit(any(), any(), any(), any(), any())).thenReturn("test-job-id")
+
DataprocSubmitter.run(
args = baseArgs.toArray,
submitter = submitter,
- envMap = Map("GCP_BIGTABLE_INSTANCE_ID" -> Option("zipline-canary-instance"),
- "GCP_PROJECT_ID" -> Option("canary-443022"))
+ envMap = Map("GCP_BIGTABLE_INSTANCE_ID" -> Option("test-instance"),
+ "GCP_PROJECT_ID" -> Option("test-project"))
)
+
+ verify(submitter).submit(
+ jobType = TypeSparkServerlessJob,
+ submissionProperties = any(),
+ jobProperties = any(),
+ files = any(),
+ any()
+ )
}Committable suggestion skipped: line range outside the PR's diff.
🤖 Prompt for AI Agents
In
cloud_gcp/src/test/scala/ai/chronon/integrations/cloud_gcp/DataprocSubmitterTest.scala
around lines 1120 to 1146, the test "test spark serverless" lacks assertions and
directly calls real GCP resources. To fix this, rename the test to a more
descriptive name like "should submit spark serverless job successfully," mock
the DataprocSubmitter to prevent actual GCP calls, and add assertions to verify
that the submitter is called with the expected arguments and behaves as
intended.
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Add this suggestion to a batch that can be applied as a single commit.
This suggestion is invalid because no changes were made to the code.
Suggestions cannot be applied while the pull request is closed.
Suggestions cannot be applied while viewing a subset of changes.
Only one suggestion per line can be applied in a batch.
Add this suggestion to a batch that can be applied as a single commit.
Applying suggestions on deleted lines is not supported.
You must change the existing code in this line in order to create a valid suggestion.
Outdated suggestions cannot be applied.
This suggestion has been applied or marked resolved.
Suggestions cannot be applied from pending reviews.
Suggestions cannot be applied on multi-line comments.
Suggestions cannot be applied while the pull request is queued to merge.
Suggestion cannot be applied right now. Please check back later.
Summary
Checklist
Summary by CodeRabbit