29373009385

Committed 14 Jul 2026 10:26PM UTC coverage: 63.842% (+0.03%) from 63.813%

Build # 29373009385

Build Type

push

github

Committed by

web-flow

Commit Message

TRUNK-6516: Fix outbox/broker issues found in events/CDC sign-off (#6298) (#6313)

Backport of #6298 to 2.9.x, adapted to the 2.9.x platform (javax
persistence / Hibernate 5 / Spring 5). Addresses findings 1, 2, 4, 5, 7
from the sign-off issue.

- Finding 1: OutboxEventRegistry now owns the scheduler lifecycle. It
  triggers OutboxTaskSchedulerInitializer.schedule()/deleteScheduledTasks()
  after scanning listener beans, guaranteeing the poller only starts once
  listeners are known. OutboxTaskSchedulerInitializer no longer implements
  SmartInitializingSingleton and no longer references OutboxEventRegistry.
- Finding 2: Use Context.loadClass() instead of Class.forName() in
  OutboxPollingTaskHandler so module-defined event types resolve.
- Finding 4: Change outbox_event.payload and completed_listeners from TEXT
  to MEDIUMTEXT in the update changelog and align the OutboxEvent entity
  columnDefinition. HibernateAdministrationDAO now treats LONGTEXT and
  MEDIUMTEXT columns as unbounded so ValidateUtil accepts large payloads.
- Finding 5: BrokerEventListenerFactory rejects wildcard/raw
  BrokerIncomingEvent at registration time instead of dead-lettering.
- Finding 7: Document on OpenmrsServiceEventAdvice that a synchronous
  @EventListener exception aborting the service call is intentional.

The 2.9.x liquibase schema-only snapshot does not exist yet, so only the
update changelog carries the MEDIUMTEXT change. Context.java javadoc link
updated to OutboxTaskSchedulerInitializer#schedule().

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Coverage Stats

19 of 27 new or added lines in 5 files covered. (70.37%)

15 existing lines in 5 files now uncovered.

24113 of 37770 relevant lines covered (63.84%)

0.64 hits per line

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

41.18

/api/src/main/java/org/openmrs/event/broker/BrokerIncomingEvent.java

/**
 * This Source Code Form is subject to the terms of the Mozilla Public License,
 * v. 2.0. If a copy of the MPL was not distributed with this file, You can
 * obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
 * the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
 *
 * Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
 * graphic logo is a trademark of OpenMRS Inc.
 */
package org.openmrs.event.broker;

import org.openmrs.event.EventPayload;
import org.openmrs.serialization.JacksonConfig;

import java.util.Map;
import java.util.Set;

/**
 * The event is published by a broker whenever there's a new message.
 * <p>
 * It needs to be used with {@link BrokerEventListener}.
 * <p>
 * You may put a condition on the listener to limit to specific source or broker e.g.
 * {@code @BrokerEventListener('my-queue', broker='my-broker")}
 * <p>
 * The payload may be either an {@code InputStream} or deserialized by a broker from JSON with 
 * {@link JacksonConfig#objectMapper()}.
 * <p>
 * In case the payload requires custom deserialization, you may extend the class and implement {@link EventPayload}.
 * <p>
 * If the payload is an {@code InputStream}, it is automatically buffered into memory on the first read.
 * If the stream size exceeds 256 kB, it spills over to a temporary file.
 * This allows multiple listeners to consume the stream's content, as each will receive a new,
 * independent {@code InputStream}. The temporary file is automatically scheduled for deletion, when the event is 
 * garbage-collected.
 * <p>
 * Values from headers are deserialized to simple values or in case of complex objects from JSON with 
 * {@link JacksonConfig#objectMapper()}.
 *
 * @since 2.9.0
 */
public class BrokerIncomingEvent<T> extends BrokerEvent<T> {
        private static final long serialVersionUID = 1L;
        
        private String source;

        public BrokerIncomingEvent() {
        }
        
        public BrokerIncomingEvent(T payload, String source) {
                super(payload);
                this.source = source;
        }

        public BrokerIncomingEvent(T payload, String source, String broker) {
                super(payload, broker);
                this.source = source;
        }
        
        public BrokerIncomingEvent(T payload, String source, String broker, Map<String, Object> headers) {
                super(payload, broker, headers);
                this.source = source;
        }

        public BrokerIncomingEvent(T payload, String source, String broker, Map<String, Object> headers, Set<String> tags) {
                super(payload, broker, headers, tags);
                this.source = source;
        }

        /**
         * The source this event came from e.g. queue name.
         * 
         * @return the source of the event
         */
        public String getSource() {
                return source;
        }

        public void setSource(String source) {
                this.source = source;
        }

}

1	/**
2	* This Source Code Form is subject to the terms of the Mozilla Public License,
3	* v. 2.0. If a copy of the MPL was not distributed with this file, You can
4	* obtain one at http://mozilla.org/MPL/2.0/. OpenMRS is also distributed under
5	* the terms of the Healthcare Disclaimer located at http://openmrs.org/license.
6	*
7	* Copyright (C) OpenMRS Inc. OpenMRS is a registered trademark and the OpenMRS
8	* graphic logo is a trademark of OpenMRS Inc.
9	*/
10	package org.openmrs.event.broker;
11
12	import org.openmrs.event.EventPayload;
13	import org.openmrs.serialization.JacksonConfig;
14
15	import java.util.Map;
16	import java.util.Set;
17
18	/**
19	* The event is published by a broker whenever there's a new message.
20	* <p>
21	* It needs to be used with {@link BrokerEventListener}.
22	* <p>
23	* You may put a condition on the listener to limit to specific source or broker e.g.
24	* {@code @BrokerEventListener('my-queue', broker='my-broker")}
25	* <p>
26	* The payload may be either an {@code InputStream} or deserialized by a broker from JSON with
27	* {@link JacksonConfig#objectMapper()}.
28	* <p>
29	* In case the payload requires custom deserialization, you may extend the class and implement {@link EventPayload}.
30	* <p>
31	* If the payload is an {@code InputStream}, it is automatically buffered into memory on the first read.
32	* If the stream size exceeds 256 kB, it spills over to a temporary file.
33	* This allows multiple listeners to consume the stream's content, as each will receive a new,
34	* independent {@code InputStream}. The temporary file is automatically scheduled for deletion, when the event is
35	* garbage-collected.
36	* <p>
37	* Values from headers are deserialized to simple values or in case of complex objects from JSON with
38	* {@link JacksonConfig#objectMapper()}.
39	*
40	* @since 2.9.0
41	*/
42	public class BrokerIncomingEvent<T> extends BrokerEvent<T> {
43	private static final long serialVersionUID = 1L;
44
45	private String source;
46
UNCOV 47	public BrokerIncomingEvent() {	×
UNCOV 48	}	×
49
50	public BrokerIncomingEvent(T payload, String source) {
51	super(payload);	1✔
52	this.source = source;	1✔
53	}	1✔
54
55	public BrokerIncomingEvent(T payload, String source, String broker) {
56	super(payload, broker);	1✔
57	this.source = source;	1✔
58	}	1✔
59
60	public BrokerIncomingEvent(T payload, String source, String broker, Map<String, Object> headers) {
61	super(payload, broker, headers);	×
62	this.source = source;	×
63	}	×
64
65	public BrokerIncomingEvent(T payload, String source, String broker, Map<String, Object> headers, Set<String> tags) {
66	super(payload, broker, headers, tags);	×
67	this.source = source;	×
68	}	×
69
70	/**
71	* The source this event came from e.g. queue name.
72	*
73	* @return the source of the event
74	*/
75	public String getSource() {
76	return source;	1✔
77	}
78
79	public void setSource(String source) {
UNCOV 80	this.source = source;	×
UNCOV 81	}	×
82
83	}

openmrs / openmrs-core / 29373009385

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