TRUNK-6516: Provide CDC mechanism with Debezium (#6151)

Author: rkorytkowski

api/src/main/java/org/openmrs/api/OpenmrsPropertyConfig.java

@@ -16,6 +16,7 @@
 import org.springframework.context.support.PropertySourcesPlaceholderConfigurer;
 import org.springframework.core.io.ClassPathResource;
 import org.springframework.core.io.FileSystemResource;
+
 import java.util.Properties;
 
 /**

api/src/main/java/org/openmrs/event/AggregatedEntityEvent.java

@@ -9,7 +9,7 @@
  */
 package org.openmrs.event;
 
-import java.util.HashSet;
+import java.util.ArrayList;
 import java.util.List;
 import java.util.Set;
 
@@ -18,21 +18,23 @@
  * <p>
  * It needs to be used together with {@link TransactionalEventAggregator}.
  * 
- * @since 2.9.x
+ * @since 2.9.0
  */
-public abstract class AggregatedEntityEvent extends BaseEvent {
+public abstract class AggregatedEntityEvent extends BaseSessionEvent {
+	private static final long serialVersionUID = 1L;
 
 	private List<EntityEvent<?>> events;
 
 	public AggregatedEntityEvent() {
+		this.events = new ArrayList<>();
 	}
 
 	public AggregatedEntityEvent(List<EntityEvent<?>> events) {
-		this(events, new HashSet<>());
+		this.events = events;
 	}
 
 	public AggregatedEntityEvent(List<EntityEvent<?>> events, Set<String> tags) {
-		super(events, tags);
+		super(tags);
 		this.events = events;
 	}
 
@@ -42,6 +44,5 @@ public List<EntityEvent<?>> getEvents() {
 
 	public void setEvents(List<EntityEvent<?>> events) {
 		this.events = events;
-		this.source = events;
 	}
 }

api/src/main/java/org/openmrs/event/BaseEvent.java

@@ -10,7 +10,6 @@
 package org.openmrs.event;
 
 import org.openmrs.event.outbox.OutboxableEvent;
-import org.springframework.context.ApplicationEvent;
 
 import java.util.HashSet;
 import java.util.Set;
@@ -20,36 +19,22 @@
  * <p>
  * Extend {@link EntityEvent} or {@link AggregatedEntityEvent} instead.
  * 
- * @since 2.9.x
+ * @since 2.9.0
  */
-class BaseEvent extends ApplicationEvent implements OutboxableEvent {
-
+class BaseEvent implements OutboxableEvent {
+	private static final long serialVersionUID = 1L;
+	
 	protected final Set<String> tags;
-	private String sessionId;
 
 	public BaseEvent() {
-		this(""); //empty for deserialization to work
-	}
-
-	public BaseEvent(Object source) {
-		this(source, new HashSet<>());
+		this.tags = new HashSet<>();
 	}
 
-	public BaseEvent(Object source, Set<String> tags) {
-		super(source);
+	public BaseEvent(Set<String> tags) {
 		this.tags = tags;
 	}
-
-
+	
 	public Set<String> getTags() {
 		return tags;
 	}
-
-	public String getSessionId() {
-		return sessionId;
-	}
-
-	public void setSessionId(String sessionId) {
-		this.sessionId = sessionId;
-	}
 }

api/src/main/java/org/openmrs/event/BaseSessionEvent.java

@@ -0,0 +1,46 @@
+/**
+ * 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;
+
+import java.util.Set;
+
+/**
+ * See {@link EntityEvent} or {@link AggregatedEntityEvent}.
+ * 
+ * @since 2.9.0
+ */
+public abstract class BaseSessionEvent extends BaseEvent {
+	private static final long serialVersionUID = 1L;
+	
+	protected String sessionId;
+
+	/**
+	 * Default constructor for deserialization.
+	 */
+	public BaseSessionEvent() {
+	}
+
+	public BaseSessionEvent(Set<String> tags) {
+		super(tags);
+	}
+	
+	public BaseSessionEvent(String sessionId, Set<String> tags) {
+		super(tags);
+		this.sessionId = sessionId;
+	}
+
+	public String getSessionId() {
+		return sessionId;
+	}
+
+	public void setSessionId(String sessionId) {
+		this.sessionId = sessionId;
+	}
+}

api/src/main/java/org/openmrs/event/CDCEvent.java

@@ -0,0 +1,164 @@
+/**
+ * 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;
+
+import org.codehaus.jackson.annotate.JsonIgnore;
+import org.jspecify.annotations.Nullable;
+import org.springframework.core.ResolvableType;
+import org.springframework.core.ResolvableTypeProvider;
+
+import java.util.Map;
+import java.util.Set;
+
+/**
+ * Issued by the CDC engine (if available, see e.g. https://github.com/openmrs/openmrs-module-debezium)
+ * <p>
+ * Please note that CDC events are usually not happening inside an application transaction rather
+ * they are fetched from DB by a separate process e.g. Debezium. For this reason it does not make sense
+ * to use {@link org.springframework.transaction.event.TransactionalEventListener} to process them.
+ * <p>
+ * If you want to guarantee delivery, please use {@link org.springframework.context.event.EventListener} 
+ * so that an event is processed in the same thread as the publisher and if the processing fails, it will be
+ * re-tried by the publisher. It is recommended only for fast enough listeners so that publisher is not
+ * blocked for too long, and it keeps up with the events queue.
+ * <p>
+ * If you combine {@link org.springframework.context.event.EventListener} with {@link org.springframework.scheduling.annotation.Async}
+ * you may run longer processing, but you lose the ability to re-try if anything fails, and in worst case you may lose the event.
+ * <p>
+ * For longer running processing with guaranteed delivery you may use {@link org.openmrs.event.outbox.OutboxEventListener}
+ * to persist events to the database and have them processed asynchronously or publish events with 
+ * {@link org.springframework.context.event.EventListener} to a message broker for asynchronous processing.
+ * The latter is more performant, and it can also be used to exchange messages with external systems.
+ * 
+ * @param <T> entityType associated with this event
+ * 
+ * @since 2.9.0
+ */
+public class CDCEvent<T> extends BaseEvent implements ResolvableTypeProvider {
+	private static final long serialVersionUID = 1L;
+	
+	private String snapshot;
+	
+	private Class<T> entityType;
+	
+	private String transactionId;
+
+	private Operation operation;
+	
+	private String tableName;
+	
+	private Map<String, Object> primaryKey;
+
+	private Map<String, Object> newState;
+	
+	private Map<String, Object> previousState;
+	
+	/**
+	 * Default constructor for deserialization.
+	 */
+	public CDCEvent() {
+	}
+	
+	public CDCEvent(Class<T> entityType) {
+		this.entityType = entityType;
+	}
+
+	public CDCEvent(Class<T> entityType, Set<String> tags) {
+		super(tags);
+		this.entityType = entityType;
+	}
+
+	public void setPrimaryKey(Map<String, Object> primaryKey) {
+		this.primaryKey = primaryKey;
+	}
+
+	public Map<String, Object> getPrimaryKey() {
+		return primaryKey;
+	}
+	
+	public boolean isCompositePrimaryKey() {
+		return primaryKey != null && primaryKey.size() > 1;
+	}
+
+	public void setEntityType(Class<T> entityType) {
+		this.entityType = entityType;
+	}
+
+	public Class<T> getEntityType() {
+		return entityType;
+	}
+
+	public void setTransactionId(String transactionId) {
+		this.transactionId = transactionId;
+	}
+
+	public String getTransactionId() {
+		return transactionId;
+	}
+
+	public void setTableName(String tableName) {
+		this.tableName = tableName;
+	}
+
+	public String getTableName() {
+		return tableName;
+	}
+
+	public void setOperation(Operation operation) {
+		this.operation = operation;
+	}
+
+	public Operation getOperation() {
+		return operation;
+	}
+
+	/**
+	 * Indicates that the event is an initial data synchronization.
+	 * 
+	 * @return true, first or last for snapshot event, false or null otherwise
+	 */
+	public String getSnapshot() {
+		return snapshot;
+	}
+
+	public void setSnapshot(String snapshot) {
+		this.snapshot = snapshot;
+	}
+
+	public Map<String, Object> getNewState() {
+		return newState;
+	}
+
+	public void setNewState(Map<String, Object> newState) {
+		this.newState = newState;
+	}
+
+	public Map<String, Object> getPreviousState() {
+		return previousState;
+	}
+
+	public void setPreviousState(Map<String, Object> previousState) {
+		this.previousState = previousState;
+	}
+	
+	public enum Operation {
+		READ, CREATE, UPDATE, DELETE, TRUNCATE 
+	}
+
+	@JsonIgnore
+	@Override
+	public @Nullable ResolvableType getResolvableType() {
+		if (entityType != null && getClass().getTypeParameters().length == 1) {
+			return ResolvableType.forClassWithGenerics(getClass(), ResolvableType.forClass(entityType));
+		} else {
+			return ResolvableType.forClass(getClass());
+		}
+	}
+}

api/src/main/java/org/openmrs/event/CDCTransactionCompletedEvent.java

@@ -0,0 +1,38 @@
+/**
+ * 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;
+
+/**
+ * Issued by the CDC engine upon completed transaction. It can be used to gather
+ * events belonging to the same transaction.
+ * 
+ * @since 2.9.0
+ */
+public class CDCTransactionCompletedEvent extends BaseEvent {
+	
+	private static final long serialVersionUID = 1L;
+	
+	private String transactionId;
+	
+	public CDCTransactionCompletedEvent() {
+	}
+	
+	public CDCTransactionCompletedEvent(String transactionId) {
+		this.transactionId = transactionId;
+	}
+	
+	public String getTransactionId() {
+		return transactionId;
+	}
+	
+	public void setTransactionId(String transactionId) {
+		this.transactionId = transactionId;
+	}
+}

api/src/main/java/org/openmrs/event/CDCTransactionStartedEvent.java

@@ -0,0 +1,37 @@
+/**
+ * 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;
+
+/**
+ * Issued by the CDC engine upon started transaction.
+ * 
+ * @since 2.9.0
+ */
+public class CDCTransactionStartedEvent extends BaseEvent {
+	
+	private static final long serialVersionUID = 1L;
+	
+	private String transactionId;
+	
+	public CDCTransactionStartedEvent() {
+	}
+	
+	public CDCTransactionStartedEvent(String transactionId) {
+		this.transactionId = transactionId;
+	}
+	
+	public String getTransactionId() {
+		return transactionId;
+	}
+	
+	public void setTransactionId(String transactionId) {
+		this.transactionId = transactionId;
+	}
+}