Hacktoberfest 2026: le issue che i maintainer hanno segnato per ottobre, aperte e adatte ai principianti. Sfoglia le issue Hacktoberfest

Topic expiry task loads the whole JDBC topic store into the heap on PostgreSQL (AMQ-6067 fix is ineffective with pgjdbc)

Aperta
  • Qualcosa non va?
#2,630 4 commenti 0 reazioni 0 assegnatari Vedi su GitHub

I maintainer di solito rispondono entro 1 giorno

@papinifrancesco ci sta già lavorando.

Dal 2/10/2026.

  • #2634 di @papinifrancesco — aperta

Valutazione

Difficoltà
4/5
Tempo stimato
3-5 giorni
Idoneità per principianti
48/100
Tipo di issue
Bug
Chiarezza
Abbastanza chiara
Stato di attività
Attiva
Stack tecnologico
java, postgresql
Ambito
backend, databases, performance

Direzione di ricerca

Start with Topic.java, JDBCMessageStore.java, and DefaultJDBCAdapter.java at the referenced expiry and recovery paths. Run the linked synthetic PostgreSQL reproduction, then compare the JDBC recovery behavior with the bounded KahaDB path and existing setMaxRows calls. Done means topic expiry no longer loads the full JDBC result set and remains bounded by maxExpirePageSize.

Scritto dal modello di indicizzazione a partire dal testo della issue.

Descrizione

Summary

With the JDBC persistence adapter, the periodic topic expiry task (Topic.expireMessagesTask, every expireMessagesPeriod, default 30 s) falls back to Topic.doBrowse(). That calls JDBCMessageStore.recover() → DefaultJDBCAdapter.doRecover(), which runs

SELECT ID, MSG FROM ACTIVEMQ_MSGS WHERE CONTAINER=? ORDER BY ID

with no setMaxRows() and no fetch size. AMQ-6067 made the recovery listener stop after maxExpirePageSize (400) messages. That is enough for drivers that stream rows, but the PostgreSQL driver reads the entire result set into memory inside executeQuery(), before the first rs.next(). On PostgreSQL, every run therefore loads every stored message of the topic (the whole durable-subscription backlog) into the heap, only to use 400 of them.

Reproduction: https://github.com/papinifrancesco/activemq-jdbc-topic-expiry-oom reproduces it on 6.2.10 and 6.3.2, with synthetic data, heap dumps and a one-click GitHub Actions workflow (details below).

Environment
  • ActiveMQ 6.2.6 when it happened; the same code is in 6.2.10 and 6.3.2 (references below)
  • JDBC persistence adapter on PostgreSQL (pgjdbc 42.7.x, commons-dbcp2), default statements and table names
  • One topic with a durable subscription whose consumer fell behind during a bulk load: ~347,000 persistent messages of ~5 KB stored
  • JVM heap ≈ 2.4 GB, default expireMessagesPeriod, no message sets a TTL
What happened

The broker stopped with java.lang.OutOfMemoryError: Java heap space. In the heap dump (Eclipse MAT, Leak Suspects), 69% of the heap (1.78 GB) was one ArrayList local to the thread ActiveMQ BrokerService[activemq] Task-614, holding 346,764 org.postgresql.core.Tuple (693,531 byte[]). Stack of that thread (line numbers from 6.2.6):

org.postgresql.core.v3.QueryExecutorImpl.processResults(QueryExecutorImpl.java:2387)
org.postgresql.core.v3.QueryExecutorImpl.execute(QueryExecutorImpl.java:372)
org.postgresql.jdbc.PgStatement.executeInternal(PgStatement.java:517)
org.postgresql.jdbc.PgPreparedStatement.executeQuery(PgPreparedStatement.java:137)
org.apache.commons.dbcp2.DelegatingPreparedStatement.executeQuery(DelegatingPreparedStatement.java:123)
org.apache.activemq.store.jdbc.adapter.DefaultJDBCAdapter.doRecover(DefaultJDBCAdapter.java:406)
org.apache.activemq.store.jdbc.JDBCMessageStore.recover(JDBCMessageStore.java:279)
org.apache.activemq.store.ProxyTopicMessageStore.recover(ProxyTopicMessageStore.java:63)
org.apache.activemq.broker.region.Topic.doBrowse(Topic.java:678)
org.apache.activemq.broker.region.Topic.lambda$new$2(Topic.java:920)
java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136)

The thread is still inside executeQuery(), receiving tuples: the listener's hasSpace() check (AMQ-6067) never gets a chance to stop it.

Reproduction (synthetic, public)

https://github.com/papinifrancesco/activemq-jdbc-topic-expiry-oom uses only the official binary distributions, a stock PostgreSQL 17 and the broker's own activemq consumer / activemq producer CLI. A GitHub Actions workflow runs four cases. Each run offers 200,000 × 5 KB persistent messages, no TTL, to a topic with one offline durable subscriber, on a 1 GB heap:

ActiveMQ Store expireMessagesPeriod Result
6.2.10 JDBC / PostgreSQL default OutOfMemoryError at ~86k stored rows, Failed to browse Topic WARN, browse caught 3× in DefaultJDBCAdapter.doRecover
6.3.2 JDBC / PostgreSQL default OutOfMemoryError at ~89k stored rows, broker exits, same stack
6.2.10 JDBC / PostgreSQL 0 all 200,000 rows stored, no OOM
6.2.10 KahaDB default no OOM

MAT on the 6.2.10 dump: thread ActiveMQ BrokerService[repro] Task-3 retains 53.8% of the heap in 54,343 org.postgresql.core.Tuple objects. Its stack is the same as above (Topic.lambda$new$2:950 → Topic.doBrowse:691 → JDBCMessageStore.recover:279 → DefaultJDBCAdapter.doRecover:406 → executeQuery). Most of the remaining heap (28%) was the durable subscription's pending cache. We capped that separately with a topic memoryLimit; it is not part of this report.

Code references (6.3.2; same code on activemq-6.2.x)
  • Topic.java:930 uses store.recoverExpired(...) only when store.getType() == StoreType.KAHADB (added by AMQ-9698). Any other store takes Topic.java:980-982: doBrowse(new InsertionCountList<>(), getMaxExpirePageSize()). On activemq-6.2.x these are lines 898 and 950.
  • DefaultJDBCAdapter.java:399-404: doRecover runs findAllMessagesStatement with no limit. The other recovery queries in the same class do limit it: setMaxRows(...) at lines 433, 593, 627 and 1068.
  • JDBCMessageStore.java:274-296: the hasSpace() check per row (AMQ-6067) runs after executeQuery() has returned.
  • Both JDBC files have the same line numbers on activemq-6.2.x.
Why the AMQ-6067 fix does not cover PostgreSQL

AMQ-6067 was reported on Oracle, whose driver streams rows (default fetch size 10), so stopping after 400 rows fixed it there. pgjdbc buffers the whole result set unless the connection is not in autocommit mode and a fetch size is set (pgjdbc docs, "Getting results based on a cursor"). doRecover sets no fetch size, and the stack above shows the driver buffering the full result set. MySQL Connector/J also buffers full result sets by default, so it is probably affected too; we have not tested it.

Expected behaviour

The expiry browse of a JDBC topic store reads at most maxExpirePageSize rows, as it effectively does with KahaDB.

Possible fixes (for discussion)
  1. Give doRecover a limit and call s.setMaxRows(limit), like the other recovery methods. doBrowse already knows max.
  2. Implement recoverExpired for the JDBC store with a bounded query on the existing indexed column (ACTIVEMQ_MSGS_EIDX), e.g. ... WHERE CONTAINER=? AND EXPIRATION > 0 AND EXPIRATION < ?. The fallback would then be unnecessary, and a topic whose messages have no TTL would return zero rows.
  3. At least, skip the browse when no stored message of the topic has EXPIRATION > 0.
Workaround

expireMessagesPeriod="0" on the topic policyEntry. Expired messages are still discarded at dispatch (PrefetchSubscription.dispatchPending). They are no longer removed while a durable subscriber is offline, though: the JDBC cleanup (doDeleteOldMessages) only deletes acknowledged rows.

The production heap dump cannot be shared, because it contains customer data. The reproduction above gives the same stack and the same retention pattern with synthetic data only.

Lingua principale
Java
Stelle
2.5k
Fork
1.5k
Merge medio
21h 32m
PR unite (30g)
87

Preparare l'ambiente

Questo progetto non fornisce container di sviluppo, Dockerfile né guida per i contributori, quindi l'ambiente è a tuo carico: parti dal suo README e consulta la nostra guida al primo contributo per i passaggi generali.

Come iniziare

  1. Leggi tutta la issue e poi la guida ai contributi del progetto.
  2. Commenta sulla issue per dire che te ne occupi tu — evita che due persone facciano lo stesso lavoro.
  3. Fai un fork del repository e lavora su un branch.
  4. Apri una pull request che faccia riferimento al numero della issue.

Altre issue di apache/activemq

Tutte le issue di apache/activemq

Issue simili

Altre issue su Java

Ricevi le nuove issue nella tua casella

Un breve riepilogo di issue GitHub adatte ai principianti.