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

89.89

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

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

import java.util.concurrent.RejectedExecutionException;
import java.util.logging.Filter;
import java.util.logging.LogRecord;
import java.util.logging.Logger;
import org.apache.logging.log4j.LogManager;
import org.apache.logging.log4j.core.LogEvent;
import org.apache.logging.log4j.core.LoggerContext;
import org.apache.logging.log4j.core.config.Configuration;
import org.apache.logging.log4j.core.config.LoggerConfig;
import org.apache.logging.log4j.core.filter.AbstractFilter;

/**
 * Drops the stack a subscriber logs while it is already shutting down. Spark prints that stack
 * through log4j, so the filter is installed there. Other warnings from these loggers stay.
 */
final class MessageDispatcherLog {

  static final String DISPATCHER = "com.google.cloud.pubsub.v1.MessageDispatcher";

  static final String DEFAULT_PROMISE =
      "io.grpc.netty.shaded.io.netty.util.concurrent.DefaultPromise";

  static final String CANCEL_RUNNABLE = "io.grpc.internal.LogExceptionRunnable";

  private static final String PERIODIC_JOB = "failed to run periodic job";

  private static final String OPERATION_COMPLETE = ".operationComplete()";

  private static final String EXECUTING_RUNNABLE = "Exception while executing runnable";

  private static final String[] LOGGER_NAMES = {DISPATCHER, DEFAULT_PROMISE, CANCEL_RUNNABLE};

  private static final Filter JUL_FILTER = MessageDispatcherLog::keep;

  private MessageDispatcherLog() {}

  static void install() {
    for (String name : LOGGER_NAMES) {
      Logger.getLogger(name).setFilter(JUL_FILTER);
    }
    installLog4j();
  }

  /** {@code false} drops the record. A shutdown rejection is the only drop. */
  static boolean keep(LogRecord record) {
    if (record == null) {
      return true;
    }
    return !drop(record.getMessage(), record.getThrown());
  }

  static boolean drop(String message, Throwable thrown) {
    if (!shutdownMessage(message) || !shutdownRejection(thrown)) {
      return false;
    }
    return true;
  }

  private static boolean shutdownMessage(String message) {
    if (message == null) {
      return false;
    }
    return PERIODIC_JOB.equals(message)
        || message.contains(OPERATION_COMPLETE)
        || message.startsWith(EXECUTING_RUNNABLE);
  }

  private static boolean shutdownRejection(Throwable thrown) {
    Throwable current = thrown;
    while (current != null) {
      if (current instanceof RejectedExecutionException) {
        String detail = current.getMessage();
        if (detail != null && (detail.contains("Shutting down") || detail.contains("Terminated"))) {
          return true;
        }
      }
      current = current.getCause();
    }
    return false;
  }

  private static void installLog4j() {
    try {
      LoggerContext ctx = (LoggerContext) LogManager.getContext(false);
      Configuration config = ctx.getConfiguration();
      if (config == null) {
        return;
      }
      for (String name : LOGGER_NAMES) {
        Log4jFilter filter = new Log4jFilter();
        filter.start();
        LoggerConfig existing = config.getLoggerConfig(name);
        if (!name.equals(existing.getName())) {
          LoggerConfig created = new LoggerConfig(name, existing.getLevel(), true);
          created.addFilter(filter);
          config.addLogger(name, created);
        } else {
          existing.addFilter(filter);
        }
      }
      ctx.updateLoggers();
    } catch (Exception | LinkageError ignored) {
      // A unit test without Spark's log4j still has the JUL filter.
    }
  }

  static final class Log4jFilter extends AbstractFilter {

    @Override
    public Result filter(LogEvent event) {
      if (event == null || event.getMessage() == null) {
        return Result.NEUTRAL;
      }
      if (drop(event.getMessage().getFormattedMessage(), event.getThrown())) {
        return Result.DENY;
      }
      return Result.NEUTRAL;
    }
  }
}

1	package io.github.juarezr.spark.pubsub.structured;
2
3	import java.util.concurrent.RejectedExecutionException;
4	import java.util.logging.Filter;
5	import java.util.logging.LogRecord;
6	import java.util.logging.Logger;
7	import org.apache.logging.log4j.LogManager;
8	import org.apache.logging.log4j.core.LogEvent;
9	import org.apache.logging.log4j.core.LoggerContext;
10	import org.apache.logging.log4j.core.config.Configuration;
11	import org.apache.logging.log4j.core.config.LoggerConfig;
12	import org.apache.logging.log4j.core.filter.AbstractFilter;
13
14	/**
15	* Drops the stack a subscriber logs while it is already shutting down. Spark prints that stack
16	* through log4j, so the filter is installed there. Other warnings from these loggers stay.
17	*/
18	final class MessageDispatcherLog {
19
20	static final String DISPATCHER = "com.google.cloud.pubsub.v1.MessageDispatcher";
21
22	static final String DEFAULT_PROMISE =
23	"io.grpc.netty.shaded.io.netty.util.concurrent.DefaultPromise";
24
25	static final String CANCEL_RUNNABLE = "io.grpc.internal.LogExceptionRunnable";
26
27	private static final String PERIODIC_JOB = "failed to run periodic job";
28
29	private static final String OPERATION_COMPLETE = ".operationComplete()";
30
31	private static final String EXECUTING_RUNNABLE = "Exception while executing runnable";
32
33	private static final String[] LOGGER_NAMES = {DISPATCHER, DEFAULT_PROMISE, CANCEL_RUNNABLE};	1 ✔
34
35	private static final Filter JUL_FILTER = MessageDispatcherLog::keep;	1 ✔
36
37	private MessageDispatcherLog() {}
38
39	static void install() {
40	for (String name : LOGGER_NAMES) {	1 ✔
41	Logger.getLogger(name).setFilter(JUL_FILTER);	1 ✔
42	}
43	installLog4j();	1 ✔
44	}	1 ✔
45
46	/** {@code false} drops the record. A shutdown rejection is the only drop. */
47	static boolean keep(LogRecord record) {
48	if (record == null) {	1 ✔
49	return true;	1 ✔
50	}
51	return !drop(record.getMessage(), record.getThrown());	1 ✔
52	}
53
54	static boolean drop(String message, Throwable thrown) {
55	if (!shutdownMessage(message) \|\| !shutdownRejection(thrown)) {	1 ✔
56	return false;	1 ✔
57	}
58	return true;	1 ✔
59	}
60
61	private static boolean shutdownMessage(String message) {
62	if (message == null) {	1 !
NEW 63	return false;	×
64	}
65	return PERIODIC_JOB.equals(message)	1 ✔
66	\|\| message.contains(OPERATION_COMPLETE)	1 ✔
67	\|\| message.startsWith(EXECUTING_RUNNABLE);	1 ✔
68	}
69
70	private static boolean shutdownRejection(Throwable thrown) {
71	Throwable current = thrown;	1 ✔
72	while (current != null) {	1 ✔
73	if (current instanceof RejectedExecutionException) {	1 ✔
74	String detail = current.getMessage();	1 ✔
75	if (detail != null && (detail.contains("Shutting down") \|\| detail.contains("Terminated"))) {	1 !
76	return true;	1 ✔
77	}
78	}
79	current = current.getCause();	1 ✔
80	}
81	return false;	1 ✔
82	}
83
84	private static void installLog4j() {
85	try {
86	LoggerContext ctx = (LoggerContext) LogManager.getContext(false);	1 ✔
87	Configuration config = ctx.getConfiguration();	1 ✔
88	if (config == null) {	1 !
NEW 89	return;	×
90	}
91	for (String name : LOGGER_NAMES) {	1 ✔
92	Log4jFilter filter = new Log4jFilter();	1 ✔
93	filter.start();	1 ✔
94	LoggerConfig existing = config.getLoggerConfig(name);	1 ✔
95	if (!name.equals(existing.getName())) {	1 ✔
96	LoggerConfig created = new LoggerConfig(name, existing.getLevel(), true);	1 ✔
97	created.addFilter(filter);	1 ✔
98	config.addLogger(name, created);	1 ✔
99	} else {	1 ✔
100	existing.addFilter(filter);	1 ✔
101	}
102	}
103	ctx.updateLoggers();	1 ✔
NEW 104	} catch (Exception \| LinkageError ignored) {	×
105	// A unit test without Spark's log4j still has the JUL filter.
106	}	1 ✔
107	}	1 ✔
108
109	static final class Log4jFilter extends AbstractFilter {	1 ✔
110
111	@Override
112	public Result filter(LogEvent event) {
113	if (event == null \|\| event.getMessage() == null) {	1 !
NEW 114	return Result.NEUTRAL;	×
115	}
116	if (drop(event.getMessage().getFormattedMessage(), event.getThrown())) {	1 ✔
117	return Result.DENY;	1 ✔
118	}
119	return Result.NEUTRAL;	1 ✔
120	}
121	}
122	}

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