Skip to content

Remove SparkSession from class to prevent it being included in the serialization process #1460

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
yalimu-g merged 11 commits into from
Jan 27, 2026
+28 −4
Merged

Remove SparkSession from class to prevent it being included in the serialization process #1460

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
11 commits
Select commit Hold shift + click to select a range
b5ef154
fix: Remove exclusion of `com.google.cloud.bigquery.connector.common`…
yalimu-g Jan 16, 2026
b5b296b
refactor: Update QueryParameterHelper to use standard Map and List in…
yalimu-g Jan 16, 2026
b91527c
Spotless apply
yalimu-g Jan 17, 2026
c7d36da
Making the sparkSession field transient
yalimu-g Jan 20, 2026
a74bb8d
Merge branch 'GoogleCloudDataproc:master' into 1455
yalimu-g Jan 20, 2026
220405e
Revert "Spotless apply"
yalimu-g Jan 20, 2026
24c4ddf
Revert "refactor: Update QueryParameterHelper to use standard Map and…
yalimu-g Jan 20, 2026
4151a6b
Revert "fix: Remove exclusion of `com.google.cloud.bigquery.connector…
yalimu-g Jan 20, 2026
ae12ec5
Merge branch '1455' of https://github.com/yalimu-g/spark-bigquery-con…
yalimu-g Jan 20, 2026
1aa61c2
Merge branch '1455' of https://github.com/yalimu-g/spark-bigquery-con…
yalimu-g Jan 20, 2026
e2060e0
Remove sparkSession from class variable
yalimu-g Jan 20, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
6 changes: 2 additions & 4 deletions ...rc/main/java/com/google/cloud/spark/bigquery/metrics/SparkBigQueryReadSessionMetrics.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -41,7 +41,6 @@ public class SparkBigQueryReadSessionMetrics extends SparkListener
private final LongAccumulator scanTimeAccumulator;
private final LongAccumulator parseTimeAccumulator;
private final String sessionId;
private final SparkSession sparkSession;
private final long timestamp;
private final DataFormat readDataFormat;
private final DataOrigin dataOrigin;
Expand All @@ -57,7 +56,6 @@ private SparkBigQueryReadSessionMetrics(
DataFormat readDataFormat,
DataOrigin dataOrigin,
long numReadStreams) {
this.sparkSession = sparkSession;
this.sessionId = sessionName;
this.timestamp = timestamp;
this.readDataFormat = readDataFormat;
Expand Down Expand Up @@ -230,12 +228,12 @@ public void onJobEnd(SparkListenerJobEnd jobEnd) {

Method buildMethod = eventBuilderClass.getDeclaredMethod("build");

sparkSession
SparkSession.active()
.sparkContext()
.listenerBus()
.post((SparkListenerEvent) buildMethod.invoke(builderInstance));

sparkSession.sparkContext().removeSparkListener(this);
SparkSession.active().sparkContext().removeSparkListener(this);
} catch (ReflectiveOperationException ex) {
logger.debug("spark.events.SparkBigQueryConnectorReadSessionEvent library not in class path");
}
Expand Down
26 changes: 26 additions & 0 deletions ...est/java/com/google/cloud/spark/bigquery/metrics/SparkBigQueryReadSessionMetricsTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -63,4 +63,30 @@ public void testReadSessionMetricsAccumulator() {
metrics.incrementScanTimeAccumulator(5000);
assertThat(metrics.getScanTime()).isEqualTo(6000);
}

@Test
public void testSerialization() throws Exception {
String sessionName = "projects/test-project/locations/us/sessions/testSession";
SparkBigQueryReadSessionMetrics metrics =
SparkBigQueryReadSessionMetrics.from(
spark,
ReadSession.newBuilder().setName(sessionName).build(),
10L,
DataFormat.ARROW,
DataOrigin.QUERY,
10L);

java.io.ByteArrayOutputStream bos = new java.io.ByteArrayOutputStream();
java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bos);
out.writeObject(metrics);
out.close();
Comment on lines +80 to +82
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jan 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To ensure resources are properly managed and to make the code more robust, it's recommended to use a try-with-resources statement for the ObjectOutputStream. This guarantees that the stream is closed even if an exception is thrown.

Suggested change
java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bos);
out.writeObject(metrics);
out.close();
try (java.io.ObjectOutputStream out = new java.io.ObjectOutputStream(bos)) {
out.writeObject(metrics);
}


java.io.ByteArrayInputStream bis = new java.io.ByteArrayInputStream(bos.toByteArray());
java.io.ObjectInputStream in = new java.io.ObjectInputStream(bis);
SparkBigQueryReadSessionMetrics deserializedMetrics =
(SparkBigQueryReadSessionMetrics) in.readObject();
Comment on lines +84 to +87
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jan 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

The ObjectInputStream is not being closed, which can lead to resource leaks in other contexts. It's best practice to use a try-with-resources statement to manage the stream's lifecycle automatically, ensuring it's always closed.

Suggested change
java.io.ByteArrayInputStream bis = new java.io.ByteArrayInputStream(bos.toByteArray());
java.io.ObjectInputStream in = new java.io.ObjectInputStream(bis);
SparkBigQueryReadSessionMetrics deserializedMetrics =
(SparkBigQueryReadSessionMetrics) in.readObject();
SparkBigQueryReadSessionMetrics deserializedMetrics;
try (java.io.ByteArrayInputStream bis = new java.io.ByteArrayInputStream(bos.toByteArray());
java.io.ObjectInputStream in = new java.io.ObjectInputStream(bis)) {
deserializedMetrics = (SparkBigQueryReadSessionMetrics) in.readObject();
}


assertThat(deserializedMetrics.getNumReadStreams()).isEqualTo(10L);
assertThat(deserializedMetrics.getBytesRead()).isEqualTo(0);
Copy link
Contributor

@gemini-code-assist gemini-code-assist bot Jan 20, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

medium

To make the test more thorough, it would be good to also assert that the other metric accumulators (rowsRead, parseTime, scanTime) are correctly initialized to 0 after deserialization. This will provide stronger guarantees about the serialization behavior.

    assertThat(deserializedMetrics.getBytesRead()).isEqualTo(0);
    assertThat(deserializedMetrics.getRowsRead()).isEqualTo(0);
    assertThat(deserializedMetrics.getParseTime()).isEqualTo(0);
    assertThat(deserializedMetrics.getScanTime()).isEqualTo(0);

}
}
Loading