juarezr / spark-streaming-google-pubsub / 68

Committed 30 Oct 2026 09:10PM UTC coverage: 79.199% (+8.1%) from 71.066%

Build # 68

Build Type

push

github

Committed by

web-flow

Commit Message

Fix: tune auto receive time to achieve the maximum throughput possible with the available heap and RSS memory in the driver (#23)

- Make the connector discover and monitor the available memory limits
for the driver in the instance
- Adapt the message rate to the available memory to avoid out-of-memory
errors and being killed by the OOM killer
- Refine cycle time estimation logic to improve the throughput
- Add a new ackMode=AfterCheckpoint option to improve memory usage and
throughput
- Add Spark v4.3 support and testing in CI

Coverage Stats

1226 of 1711 branches covered (71.65%)

Branch coverage included in aggregate %.

862 of 1033 new or added lines in 11 files covered. (83.45%)

4 existing lines in 1 file now uncovered.

2315 of 2760 relevant lines covered (83.88%)

0.84 hits per line

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

56.49

/src/main/java/io/github/juarezr/spark/pubsub/common/Into.java

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

import java.time.Duration;
import java.time.Instant;
import java.time.OffsetDateTime;
import java.util.Locale;

public final class Into {
  static final String NONE = "none";

  public static String timeIso(Long millis) {
    if (millis == null) {
      return NONE;
    }
    return timeIso(millis.longValue());
  }

  public static String timeIso(long millis) {
    if (millis == Long.MIN_VALUE) {
      return NONE;
    }
    return Instant.ofEpochMilli(millis).toString();
  }

  public static String elapsed(Duration duration) {
    return elapsed(duration, "-");
  }

  public static String elapsed(Duration duration, String alternative) {
    if (duration == null) {
      return alternative;
    }
    final long ms = duration.toMillis();
    return elapsedMs(ms);
  }

  /** Formats a duration given in nanoseconds (not milliseconds). */
  public static String elapsed(long nanos) {
    return elapsed(Duration.ofNanos(nanos));
  }

  private static String elapsedMs(long millis) {
    if (millis > 600000L) {
      return String.format("%d:%02d", millis / 60000L, (millis % 60000L) / 1000L);
    }
    if (millis > 99999L) {
      return String.format("%.1fm", millis / 60000f).replace(".0", "");
    }
    return millis + "ms";
  }

  public static String abrevBytes(long size) {
    if (size <= 0) {
      return "0";
    }
    if (size < 1024) {
      return size + "B";
    }
    if (size < 1024 * 1024) {
      return (size / 1024) + "KB";
    }
    if (size < 1024 * 1024 * 1024) {
      return (size / 1024 / 1024) + "MB";
    }
    if (size < 1024 * 1024 * 1024 * 1024) {
      return (size / 1024 / 1024 / 1024) + "GB";
    }
    if (size < 1024 * 1024 * 1024 * 1024) {
      return (size / 1024 / 1024 / 1024 / 1024) + "TB";
    }
    return size + "B";
  }

  public static String abrevCount(long size) {
    if ((size >= 1024) && (size < 1024 * 1024)) {
      return (size / 1024) + "k";
    }
    if (size < 1024 * 1024 * 1024) {
      return (size / 1024 / 1024) + "m";
    }
    if (size < 1024 * 1024 * 1024 * 1024) {
      return (size / 1024 / 1024 / 1024) + "g";
    }
    if (size < 1024 * 1024 * 1024 * 1024) {
      return (size / 1024 / 1024 / 1024 / 1024) + "t";
    }
    return size + "";
  }

  public static long parseSize(String option, String raw) {
    if (raw == null || raw.isBlank()) {
      return 0L;
    }
    String value = raw.trim().toLowerCase(Locale.ROOT);
    long multiplier = 1L;
    char suffix = value.charAt(value.length() - 1);
    if (suffix == 'k' || suffix == 'm' || suffix == 'g') {
      value = value.substring(0, value.length() - 1);
      multiplier = suffix == 'k' ? 1024L : suffix == 'm' ? 1024L * 1024L : 1024L * 1024L * 1024L;
    }
    try {
      return Math.multiplyExact(Long.parseLong(value), multiplier);
    } catch (ArithmeticException | NumberFormatException e) {
      throw new IllegalArgumentException(
          "Invalid " + option + " '" + raw + "'. Use bytes or a k, m, or g suffix.", e);
    }
  }

  public static Duration parseDuration(String option, String raw) {
    if (raw == null || raw.isBlank()) {
      throw new IllegalArgumentException(option + " must not be blank");
    }
    String value = raw.trim().toLowerCase(Locale.ROOT);
    try {
      if (value.endsWith("ms")) {
        return Duration.ofMillis(Long.parseLong(value.substring(0, value.length() - 2)));
      }
      if (value.endsWith("s")) {
        return Duration.ofSeconds(Long.parseLong(value.substring(0, value.length() - 1)));
      }
      if (value.endsWith("m")) {
        return Duration.ofMinutes(Long.parseLong(value.substring(0, value.length() - 1)));
      }
      return Duration.ofSeconds(Long.parseLong(value));
    } catch (ArithmeticException | NumberFormatException e) {
      throw new IllegalArgumentException(
          "Invalid " + option + ": '" + raw + "'. Use a number with ms, s, or m.", e);
    }
  }

  public static Instant parseInstant(String option, String raw) {
    String value = raw == null ? "" : raw.trim();
    try {
      if (!value.isEmpty() && value.chars().allMatch(Character::isDigit)) {
        return Instant.ofEpochMilli(Long.parseLong(value));
      }
      return OffsetDateTime.parse(value).toInstant();
    } catch (Exception e) {
      throw new IllegalArgumentException(
          "Invalid " + option + " '" + raw + "'. Use epoch milliseconds or RFC-3339 with Z/offset.",
          e);
    }
  }
}

1	package io.github.juarezr.spark.pubsub.common;
2
3	import java.time.Duration;
4	import java.time.Instant;
5	import java.time.OffsetDateTime;
6	import java.util.Locale;
7
8	public final class Into {	×
9	static final String NONE = "none";
10
11	public static String timeIso(Long millis) {
12	if (millis == null) {	1 !
13	return NONE;	1 ✔
14	}
15	return timeIso(millis.longValue());	×
16	}
17
18	public static String timeIso(long millis) {
19	if (millis == Long.MIN_VALUE) {	1 ✔
20	return NONE;	1 ✔
21	}
22	return Instant.ofEpochMilli(millis).toString();	1 ✔
23	}
24
25	public static String elapsed(Duration duration) {
26	return elapsed(duration, "-");	1 ✔
27	}
28
29	public static String elapsed(Duration duration, String alternative) {
30	if (duration == null) {	1 ✔
31	return alternative;	1 ✔
32	}
33	final long ms = duration.toMillis();	1 ✔
34	return elapsedMs(ms);	1 ✔
35	}
36
37	/** Formats a duration given in nanoseconds (not milliseconds). */
38	public static String elapsed(long nanos) {
39	return elapsed(Duration.ofNanos(nanos));	1 ✔
40	}
41
42	private static String elapsedMs(long millis) {
43	if (millis > 600000L) {	1 ✔
44	return String.format("%d:%02d", millis / 60000L, (millis % 60000L) / 1000L);	1 ✔
45	}
46	if (millis > 99999L) {	1 ✔
47	return String.format("%.1fm", millis / 60000f).replace(".0", "");	1 ✔
48	}
49	return millis + "ms";	1 ✔
50	}
51
52	public static String abrevBytes(long size) {
53	if (size <= 0) {	1 !
54	return "0";	×
55	}
56	if (size < 1024) {	1 !
57	return size + "B";	×
58	}
59	if (size < 1024 * 1024) {	1 !
60	return (size / 1024) + "KB";	×
61	}
62	if (size < 1024 * 1024 * 1024) {	1 !
63	return (size / 1024 / 1024) + "MB";	1 ✔
64	}
65	if (size < 1024 * 1024 * 1024 * 1024) {	×
66	return (size / 1024 / 1024 / 1024) + "GB";	×
67	}
68	if (size < 1024 * 1024 * 1024 * 1024) {	×
69	return (size / 1024 / 1024 / 1024 / 1024) + "TB";	×
70	}
71	return size + "B";	×
72	}
73
74	public static String abrevCount(long size) {
UNCOV 75	if ((size >= 1024) && (size < 1024 * 1024)) {	×
UNCOV 76	return (size / 1024) + "k";	×
77	}
UNCOV 78	if (size < 1024 * 1024 * 1024) {	×
UNCOV 79	return (size / 1024 / 1024) + "m";	×
80	}
81	if (size < 1024 * 1024 * 1024 * 1024) {	×
82	return (size / 1024 / 1024 / 1024) + "g";	×
83	}
84	if (size < 1024 * 1024 * 1024 * 1024) {	×
85	return (size / 1024 / 1024 / 1024 / 1024) + "t";	×
86	}
87	return size + "";	×
88	}
89
90	public static long parseSize(String option, String raw) {
91	if (raw == null \|\| raw.isBlank()) {	1 !
92	return 0L;	×
93	}
94	String value = raw.trim().toLowerCase(Locale.ROOT);	1 ✔
95	long multiplier = 1L;	1 ✔
96	char suffix = value.charAt(value.length() - 1);	1 ✔
97	if (suffix == 'k' \|\| suffix == 'm' \|\| suffix == 'g') {	1 !
98	value = value.substring(0, value.length() - 1);	1 ✔
99	multiplier = suffix == 'k' ? 1024L : suffix == 'm' ? 1024L * 1024L : 1024L * 1024L * 1024L;	1 !
100	}
101	try {
102	return Math.multiplyExact(Long.parseLong(value), multiplier);	1 ✔
103	} catch (ArithmeticException \| NumberFormatException e) {	×
104	throw new IllegalArgumentException(	×
105	"Invalid " + option + " '" + raw + "'. Use bytes or a k, m, or g suffix.", e);
106	}
107	}
108
109	public static Duration parseDuration(String option, String raw) {
110	if (raw == null \|\| raw.isBlank()) {	1 !
111	throw new IllegalArgumentException(option + " must not be blank");	×
112	}
113	String value = raw.trim().toLowerCase(Locale.ROOT);	1 ✔
114	try {
115	if (value.endsWith("ms")) {	1 ✔
116	return Duration.ofMillis(Long.parseLong(value.substring(0, value.length() - 2)));	1 ✔
117	}
118	if (value.endsWith("s")) {	1 ✔
119	return Duration.ofSeconds(Long.parseLong(value.substring(0, value.length() - 1)));	1 ✔
120	}
121	if (value.endsWith("m")) {	1 !
122	return Duration.ofMinutes(Long.parseLong(value.substring(0, value.length() - 1)));	×
123	}
124	return Duration.ofSeconds(Long.parseLong(value));	×
125	} catch (ArithmeticException \| NumberFormatException e) {	1 ✔
126	throw new IllegalArgumentException(	1 ✔
127	"Invalid " + option + ": '" + raw + "'. Use a number with ms, s, or m.", e);
128	}
129	}
130
131	public static Instant parseInstant(String option, String raw) {
132	String value = raw == null ? "" : raw.trim();	1 !
133	try {
134	if (!value.isEmpty() && value.chars().allMatch(Character::isDigit)) {	1 !
135	return Instant.ofEpochMilli(Long.parseLong(value));	1 ✔
136	}
137	return OffsetDateTime.parse(value).toInstant();	1 ✔
138	} catch (Exception e) {	1 ✔
139	throw new IllegalArgumentException(	1 ✔
140	"Invalid " + option + " '" + raw + "'. Use epoch milliseconds or RFC-3339 with Z/offset.",
141	e);
142	}
143	}
144	}

juarezr / spark-streaming-google-pubsub / 68

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