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

78.0

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

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

import io.github.juarezr.spark.pubsub.common.Into;
import java.io.Serializable;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Objects;

/** A Pub/Sub message pulled into Spark, including the ack id for later acknowledgement. */
final class PulledMessage implements Serializable {

  private static final long serialVersionUID = 4367024244L;

  private final String messageId;
  private final byte[] data;
  private final Map<String, String> attributes;
  private final long publishTimeMillis;
  private final String orderingKey;
  private final String ackId;

  PulledMessage(
      String messageId,
      byte[] data,
      Map<String, String> attributes,
      long publishTimeMillis,
      String orderingKey,
      String ackId) {
    this.messageId = messageId;
    this.data = data == null ? new byte[0] : data;
    this.attributes =
        attributes == null
            ? Collections.emptyMap()
            : Collections.unmodifiableMap(new HashMap<>(attributes));
    this.publishTimeMillis = publishTimeMillis;
    this.orderingKey = orderingKey == null ? "" : orderingKey;
    this.ackId = Objects.requireNonNull(ackId, "ackId");
  }

  @Override
  public String toString() {
    return messageId + "@" + Into.timeIso(publishTimeMillis);
  }

  String messageId() {
    return messageId;
  }

  byte[] data() {
    return data;
  }

  Map<String, String> attributes() {
    return attributes;
  }

  long publishTimeMillis() {
    return publishTimeMillis;
  }

  String orderingKey() {
    return orderingKey;
  }

  String ackId() {
    return ackId;
  }

  static List<String> ackIds(List<PulledMessage> messages) {
    if (messages == null || messages.isEmpty()) {
      return List.of();
    }
    List<String> ids = new ArrayList<>(messages.size());
    for (PulledMessage message : messages) {
      ids.add(message.ackId());
    }
    return ids;
  }

  static long payloadBytes(List<PulledMessage> messages) {
    if (messages == null || messages.isEmpty()) {
      return 0L;
    }
    long bytes = 0L;
    for (PulledMessage message : messages) {
      bytes += message.data().length;
    }
    return bytes;
  }
}

1	package io.github.juarezr.spark.pubsub.structured;
2
3	import io.github.juarezr.spark.pubsub.common.Into;
4	import java.io.Serializable;
5	import java.util.ArrayList;
6	import java.util.Collections;
7	import java.util.HashMap;
8	import java.util.List;
9	import java.util.Map;
10	import java.util.Objects;
11
12	/** A Pub/Sub message pulled into Spark, including the ack id for later acknowledgement. */
13	final class PulledMessage implements Serializable {
14
15	private static final long serialVersionUID = 4367024244L;
16
17	private final String messageId;
18	private final byte[] data;
19	private final Map<String, String> attributes;
20	private final long publishTimeMillis;
21	private final String orderingKey;
22	private final String ackId;
23
24	PulledMessage(
25	String messageId,
26	byte[] data,
27	Map<String, String> attributes,
28	long publishTimeMillis,
29	String orderingKey,
30	String ackId) {	1 ✔
31	this.messageId = messageId;	1 ✔
32	this.data = data == null ? new byte[0] : data;	1 !
33	this.attributes =	1 ✔
34	attributes == null	1 !
35	? Collections.emptyMap()	×
36	: Collections.unmodifiableMap(new HashMap<>(attributes));	1 ✔
37	this.publishTimeMillis = publishTimeMillis;	1 ✔
38	this.orderingKey = orderingKey == null ? "" : orderingKey;	1 !
39	this.ackId = Objects.requireNonNull(ackId, "ackId");	1 ✔
40	}	1 ✔
41
42	@Override
43	public String toString() {
NEW 44	return messageId + "@" + Into.timeIso(publishTimeMillis);	×
45	}
46
47	String messageId() {
48	return messageId;	1 ✔
49	}
50
51	byte[] data() {
52	return data;	1 ✔
53	}
54
55	Map<String, String> attributes() {
56	return attributes;	×
57	}
58
59	long publishTimeMillis() {
60	return publishTimeMillis;	1 ✔
61	}
62
63	String orderingKey() {
64	return orderingKey;	×
65	}
66
67	String ackId() {
68	return ackId;	1 ✔
69	}
70
71	static List<String> ackIds(List<PulledMessage> messages) {
72	if (messages == null \|\| messages.isEmpty()) {	1 !
73	return List.of();	×
74	}
75	List<String> ids = new ArrayList<>(messages.size());	1 ✔
76	for (PulledMessage message : messages) {	1 ✔
77	ids.add(message.ackId());	1 ✔
78	}	1 ✔
79	return ids;	1 ✔
80	}
81
82	static long payloadBytes(List<PulledMessage> messages) {
83	if (messages == null \|\| messages.isEmpty()) {	1 !
84	return 0L;	1 ✔
85	}
86	long bytes = 0L;	1 ✔
87	for (PulledMessage message : messages) {	1 ✔
88	bytes += message.data().length;	1 ✔
89	}	1 ✔
90	return bytes;	1 ✔
91	}
92	}

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