juarezr / spark-streaming-google-pubsub / 60

Committed 02 Oct 2026 07:10AM UTC coverage: 70.532% (+4.3%) from 66.187%

Build # 60

Build Type

push

github

Committed by

web-flow

Commit Message

fix: implement connection error handling to halt on non-recoverable configuration errors (#21)

- fix: implement connection error handling to halt on non-recoverable
configuration errors
- fix: validation logic to ignore AUTO placeholders for receiveTime and
processingTime
- test: add parameterized tests for schema and metadata modes
- test: add unit tests for dynamic and mixed schema modes
- refactor: removed dead code from the StreamingPull migration leftovers

Coverage Stats

764 of 1196 branches covered (63.88%)

Branch coverage included in aggregate %.

59 of 115 new or added lines in 9 files covered. (51.3%)

2 existing lines in 2 files now uncovered.

1608 of 2167 relevant lines covered (74.2%)

0.74 hits per line

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

20.34

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

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

import com.google.cloud.pubsub.v1.SchemaServiceClient;
import com.google.cloud.pubsub.v1.SchemaServiceSettings;
import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
import com.google.cloud.pubsub.v1.TopicAdminClient;
import com.google.cloud.pubsub.v1.TopicAdminSettings;
import com.google.pubsub.v1.Encoding;
import com.google.pubsub.v1.Schema;
import com.google.pubsub.v1.SchemaSettings;
import com.google.pubsub.v1.Topic;
import io.github.juarezr.spark.pubsub.config.PubSubConfig;
import java.io.IOException;
import org.apache.spark.sql.types.StructType;

/** Loads a topic's Avro/JSON schema from Pub/Sub Schema Service. */
final class PubSubTopicSchema {
  private PubSubTopicSchema() {}

  static StructType fetchPayloadSchema(PubSubConfig config) {
    try {
      return fetch(config);
    } catch (IOException e) {
      throw new IllegalStateException("Failed to load Pub/Sub topic schema", e);
    }
  }

  static StructType fromTopicSchema(Encoding encoding, Schema schema) {
    if (encoding != Encoding.JSON) {
      throw new IllegalArgumentException(
          "Unsupported schema encoding "
              + encoding
              + "; only JSON is supported (BINARY and protobuf are not)");
    }
    if (schema == null || schema.getType() != Schema.Type.AVRO) {
      Schema.Type type = schema == null ? Schema.Type.TYPE_UNSPECIFIED : schema.getType();
      throw new IllegalArgumentException(
          "Unsupported schema type " + type + "; only AVRO is supported");
    }
    return PubSubAvroSchemaMapper.toStructType(schema.getDefinition());
  }

  private static StructType fetch(PubSubConfig config) throws IOException {
    PubSubCredentialsProvider credentialsProvider =
        new PubSubCredentialsProvider(config.credentialsFile().orElse(null));
    String topicPath = resolveTopicPath(config, credentialsProvider);
    Topic topic = getTopic(config, credentialsProvider, topicPath);
    if (!topic.hasSchemaSettings()) {
      throw new IllegalArgumentException(
          "Topic "
              + topicPath
              + " has no schema; schemaMode=dynamic/mixed requires a topic schema");
    }
    SchemaSettings settings = topic.getSchemaSettings();
    Schema schema = getSchema(config, credentialsProvider, settings.getSchema());
    return fromTopicSchema(settings.getEncoding(), schema);
  }

  private static String resolveTopicPath(
      PubSubConfig config, PubSubCredentialsProvider credentialsProvider) throws IOException {
    if (config.topicPath().isPresent()) {
      return config.topicPath().get();
    }
    SubscriptionAdminSettings.Builder builder = SubscriptionAdminSettings.newBuilder();
    try (PubSubEmulator emulator = config.emulatorHost().map(PubSubEmulator::new).orElse(null)) {
      if (emulator != null) {
        emulator.configureSubscriptionAdmin(builder);
      } else {
        builder.setCredentialsProvider(credentialsProvider.getProvider());
      }
      try (SubscriptionAdminClient admin = SubscriptionAdminClient.create(builder.build())) {
        return admin.getSubscription(config.subscriptionPath()).getTopic();
      }
    }
  }

  private static Topic getTopic(
      PubSubConfig config, PubSubCredentialsProvider credentialsProvider, String topicPath)
      throws IOException {
    TopicAdminSettings.Builder builder = TopicAdminSettings.newBuilder();
    try (PubSubEmulator emulator = config.emulatorHost().map(PubSubEmulator::new).orElse(null)) {
      if (emulator != null) {
        emulator.configureTopicAdmin(builder);
      } else {
        builder.setCredentialsProvider(credentialsProvider.getProvider());
      }
      try (TopicAdminClient admin = TopicAdminClient.create(builder.build())) {
        return admin.getTopic(topicPath);
      }
    }
  }

  private static Schema getSchema(
      PubSubConfig config, PubSubCredentialsProvider credentialsProvider, String schemaName)
      throws IOException {
    SchemaServiceSettings.Builder builder = SchemaServiceSettings.newBuilder();
    try (PubSubEmulator emulator = config.emulatorHost().map(PubSubEmulator::new).orElse(null)) {
      if (emulator != null) {
        emulator.configureSchema(builder);
      } else {
        builder.setCredentialsProvider(credentialsProvider.getProvider());
      }
      try (SchemaServiceClient client = SchemaServiceClient.create(builder.build())) {
        return client.getSchema(schemaName);
      }
    }
  }
}

1	package io.github.juarezr.spark.pubsub.structured;
2
3	import com.google.cloud.pubsub.v1.SchemaServiceClient;
4	import com.google.cloud.pubsub.v1.SchemaServiceSettings;
5	import com.google.cloud.pubsub.v1.SubscriptionAdminClient;
6	import com.google.cloud.pubsub.v1.SubscriptionAdminSettings;
7	import com.google.cloud.pubsub.v1.TopicAdminClient;
8	import com.google.cloud.pubsub.v1.TopicAdminSettings;
9	import com.google.pubsub.v1.Encoding;
10	import com.google.pubsub.v1.Schema;
11	import com.google.pubsub.v1.SchemaSettings;
12	import com.google.pubsub.v1.Topic;
13	import io.github.juarezr.spark.pubsub.config.PubSubConfig;
14	import java.io.IOException;
15	import org.apache.spark.sql.types.StructType;
16
17	/** Loads a topic's Avro/JSON schema from Pub/Sub Schema Service. */
18	final class PubSubTopicSchema {
19	private PubSubTopicSchema() {}
20
21	static StructType fetchPayloadSchema(PubSubConfig config) {
22	try {
23	return fetch(config);	×
24	} catch (IOException e) {	×
25	throw new IllegalStateException("Failed to load Pub/Sub topic schema", e);	×
26	}
27	}
28
29	static StructType fromTopicSchema(Encoding encoding, Schema schema) {
30	if (encoding != Encoding.JSON) {	1 ✔
31	throw new IllegalArgumentException(	1 ✔
32	"Unsupported schema encoding "
33	+ encoding
34	+ "; only JSON is supported (BINARY and protobuf are not)");
35	}
36	if (schema == null \|\| schema.getType() != Schema.Type.AVRO) {	1 !
37	Schema.Type type = schema == null ? Schema.Type.TYPE_UNSPECIFIED : schema.getType();	1 !
38	throw new IllegalArgumentException(	1 ✔
39	"Unsupported schema type " + type + "; only AVRO is supported");
40	}
41	return PubSubAvroSchemaMapper.toStructType(schema.getDefinition());	1 ✔
42	}
43
44	private static StructType fetch(PubSubConfig config) throws IOException {
NEW 45	PubSubCredentialsProvider credentialsProvider =	×
46	new PubSubCredentialsProvider(config.credentialsFile().orElse(null));	×
NEW 47	String topicPath = resolveTopicPath(config, credentialsProvider);	×
NEW 48	Topic topic = getTopic(config, credentialsProvider, topicPath);	×
49	if (!topic.hasSchemaSettings()) {	×
50	throw new IllegalArgumentException(	×
51	"Topic "
52	+ topicPath
53	+ " has no schema; schemaMode=dynamic/mixed requires a topic schema");
54	}
55	SchemaSettings settings = topic.getSchemaSettings();	×
NEW 56	Schema schema = getSchema(config, credentialsProvider, settings.getSchema());	×
57	return fromTopicSchema(settings.getEncoding(), schema);	×
58	}
59
60	private static String resolveTopicPath(
61	PubSubConfig config, PubSubCredentialsProvider credentialsProvider) throws IOException {
62	if (config.topicPath().isPresent()) {	×
63	return config.topicPath().get();	×
64	}
65	SubscriptionAdminSettings.Builder builder = SubscriptionAdminSettings.newBuilder();	×
66	try (PubSubEmulator emulator = config.emulatorHost().map(PubSubEmulator::new).orElse(null)) {	×
67	if (emulator != null) {	×
68	emulator.configureSubscriptionAdmin(builder);	×
69	} else {
NEW 70	builder.setCredentialsProvider(credentialsProvider.getProvider());	×
71	}
72	try (SubscriptionAdminClient admin = SubscriptionAdminClient.create(builder.build())) {	×
73	return admin.getSubscription(config.subscriptionPath()).getTopic();	×
74	}
75	}
76	}
77
78	private static Topic getTopic(
79	PubSubConfig config, PubSubCredentialsProvider credentialsProvider, String topicPath)
80	throws IOException {
81	TopicAdminSettings.Builder builder = TopicAdminSettings.newBuilder();	×
82	try (PubSubEmulator emulator = config.emulatorHost().map(PubSubEmulator::new).orElse(null)) {	×
83	if (emulator != null) {	×
84	emulator.configureTopicAdmin(builder);	×
85	} else {
NEW 86	builder.setCredentialsProvider(credentialsProvider.getProvider());	×
87	}
88	try (TopicAdminClient admin = TopicAdminClient.create(builder.build())) {	×
89	return admin.getTopic(topicPath);	×
90	}
91	}
92	}
93
94	private static Schema getSchema(
95	PubSubConfig config, PubSubCredentialsProvider credentialsProvider, String schemaName)
96	throws IOException {
97	SchemaServiceSettings.Builder builder = SchemaServiceSettings.newBuilder();	×
98	try (PubSubEmulator emulator = config.emulatorHost().map(PubSubEmulator::new).orElse(null)) {	×
99	if (emulator != null) {	×
100	emulator.configureSchema(builder);	×
101	} else {
NEW 102	builder.setCredentialsProvider(credentialsProvider.getProvider());	×
103	}
104	try (SchemaServiceClient client = SchemaServiceClient.create(builder.build())) {	×
105	return client.getSchema(schemaName);	×
106	}
107	}
108	}
109	}

juarezr / spark-streaming-google-pubsub / 60

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