juarezr / spark-streaming-google-pubsub / 57

Committed 01 Oct 2026 07:02AM UTC coverage: 66.187% (-2.3%) from 68.457%

Build # 57

Build Type

push

github

Committed by

web-flow

Commit Message

fix: update AutoBatchTime logic for receiveTime estimation with write-first approach (#20)

- fix: update AutoBatchTime logic for receiveTime estimation with
write-first approach
- feat: add processingTime configuration to PubSubConfig to hint batch
interval on AutoBatchTime
- fix: remove unneded cap on writeBudget for AutoBatchTime

Coverage Stats

689 of 1169 branches covered (58.94%)

Branch coverage included in aggregate %.

232 of 354 new or added lines in 12 files covered. (65.54%)

15 existing lines in 1 file now uncovered.

1517 of 2164 relevant lines covered (70.1%)

0.7 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

0.0

/src/main/java/io/github/juarezr/spark/pubsub/structured/PubSubScan.java

package io.github.juarezr.spark.pubsub.structured;

import io.github.juarezr.spark.pubsub.config.PubSubConfig;
import org.apache.spark.sql.connector.read.Scan;
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
import org.apache.spark.sql.types.StructType;

/** Scan that exposes a Pub/Sub {@link MicroBatchStream}. */
final class PubSubScan implements Scan {
  private final PubSubConfig config;
  private final StructType readSchema;

  PubSubScan(PubSubConfig config, StructType readSchema) {
    this.config = config;
    this.readSchema = readSchema;
  }

  @Override
  public String toString() {
    return description();
  }

  @Override
  public StructType readSchema() {
    return readSchema;
  }

  @Override
  public String description() {
    return "Google Pub/Sub scan: " + config.subscriptionPath();
  }

  @Override
  public MicroBatchStream toMicroBatchStream(String checkpointLocation) {
    return new PubSubMicroBatchStream(config, config.numWriters(), readSchema);
  }
}

1	package io.github.juarezr.spark.pubsub.structured;
2
3	import io.github.juarezr.spark.pubsub.config.PubSubConfig;
4	import org.apache.spark.sql.connector.read.Scan;
5	import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
6	import org.apache.spark.sql.types.StructType;
7
8	/** Scan that exposes a Pub/Sub {@link MicroBatchStream}. */
9	final class PubSubScan implements Scan {
10	private final PubSubConfig config;
11	private final StructType readSchema;
12
13	PubSubScan(PubSubConfig config, StructType readSchema) {	×
14	this.config = config;	×
15	this.readSchema = readSchema;	×
16	}	×
17
18	@Override
19	public String toString() {
NEW 20	return description();	×
21	}
22
23	@Override
24	public StructType readSchema() {
25	return readSchema;	×
26	}
27
28	@Override
29	public String description() {
30	return "Google Pub/Sub scan: " + config.subscriptionPath();	×
31	}
32
33	@Override
34	public MicroBatchStream toMicroBatchStream(String checkpointLocation) {
35	return new PubSubMicroBatchStream(config, config.numWriters(), readSchema);	×
36	}
37	}

juarezr / spark-streaming-google-pubsub / 57

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous