Make ceSource accessible to calling application

CHANGELOG.md

@@ -6,6 +6,12 @@ This project adheres to [Semantic Versioning](http://semver.org/).
 
 The format is based on [Keep a Changelog](http://keepachangelog.com/).
 
+## Version 4.1.1 - TBD
+
+### Added
+
+- Support overriding the CloudEvent `source` suffix per message via the `eventhub.btp.subaccountId` header, for multitenant scenarios where the tenant ID differs from the subaccount ID registered as the UCL system ID
+
 ## Version 4.1.0 - 2026-03-09
 
 ### Added

README.md

@@ -29,6 +29,18 @@ The usage of CAP Java plugins is described in the [CAP Java Documentation](https
 
 The latest version can be found in the [changelog](./CHANGELOG.md) or in the [Maven Central Repository](https://central.sonatype.com/artifact/com.sap.cds/cds-feature-event-hub/versions).
 
+### Overriding the CloudEvent source suffix
+
+In multitenant scenarios the CloudEvent `source` is composed of the configured `ceSource` and, by default, the current tenant ID. If the tenant ID differs from the subaccount ID registered as the UCL system ID, an application can override the suffix per message by setting the `eventhub.btp.subaccountId` header when emitting the event:
+
+```java
+Map<String, Object> headers = new HashMap<>();
+headers.put("eventhub.btp.subaccountId", subaccountId);
+// emit the event with these headers
+```
+
+The header is consumed by the plugin and is not included in the outgoing message. If it is absent or blank, the tenant ID is used as before.
+
 ## Support, Feedback, Contributing
 
 This project is open to feature requests/suggestions, bug reports etc. via [GitHub issues](https://github.com/cap-java/cds-feature-event-hub/issues). Contribution and feedback are encouraged and always welcome. For more information about how to contribute, the project structure, as well as additional contribution information, see our [Contribution Guidelines](CONTRIBUTING.md).

cds-feature-event-hub/src/main/java/com/sap/cds/feature/messaging/eventhub/service/EventHubMessagingService.java

@@ -32,6 +32,7 @@ public class EventHubMessagingService extends AbstractMessagingService {
 	private static final Logger logger = LoggerFactory.getLogger(EventHubMessagingService.class);
 	public static final String CE_SOURCE = "ceSource";
 	public static final String SYSTEM_ID = "systemId";
+	public static final String KEY_SUBACCOUNT_ID = "eventhub.btp.subaccountId";
 
 	private final String ceSource;
 	private final String systemId;
@@ -151,16 +152,14 @@ protected void emitTopicMessage(String topic, TopicMessageEventContext context)
 			if (ceSource == null) {
 				throw new ErrorStatusException(EventHubErrorStatuses.EVENT_HUB_EMIT_MISSING_CE_SOURCE);
 			}
-
-			headers.put(CloudEventUtils.KEY_SOURCE, ceSource + tenant);
+			headers.put(CloudEventUtils.KEY_SOURCE, ceSource + resolveSourceSuffix(headers, tenant));
 		} else {
 			if (systemId == null) {
 				throw new ErrorStatusException(EventHubErrorStatuses.EVENT_HUB_EMIT_MISSING_SYSTEM_ID);
 			}
 			if (ceSource == null) {
 				throw new ErrorStatusException(EventHubErrorStatuses.EVENT_HUB_EMIT_MISSING_CE_SOURCE);
 			}
-
 			headers.put(CloudEventUtils.KEY_SOURCE, ceSource + systemId);
 		}
 
@@ -172,6 +171,13 @@ protected void emitTopicMessage(String topic, TopicMessageEventContext context)
 		}
 	}
 
+	// Allows the application to use the BTP subaccount ID as the ceSource suffix at the application level.
+	// Relevant when the tenant ID differs from the subaccount ID registered as the UCL system ID.
+	String resolveSourceSuffix(Map<String, Object> headers, String tenant) {
+		String subaccountId = (String) headers.remove(KEY_SUBACCOUNT_ID);
+		return subaccountId != null ? subaccountId : tenant;
+	}
+
 	private String getTenant(EventContext context) {
 		String tenant = context.getUserInfo().getTenant();
 

cds-feature-event-hub/src/test/java/com/sap/cds/feature/messaging/eventhub/service/EventHubMessagingServiceTest.java

@@ -2,11 +2,14 @@
 
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertNull;
 
 import java.util.HashMap;
 import java.util.Map;
 
 import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
 import com.sap.cds.feature.messaging.eventhub.utils.EventHubErrorStatuses;
 import com.sap.cds.services.environment.CdsProperties;
@@ -16,9 +19,11 @@
 import com.sap.cds.services.runtime.CdsRuntimeConfigurer;
 
 class EventHubMessagingServiceTest {
+	private EventHubMessagingService svc;
+	private CdsRuntime runtime;
 
-	@Test
-	void testEmit_TenantNotSupported() {
+	@BeforeEach
+	public void setUp() throws Exception {
 		CdsProperties properties = new CdsProperties();
 		CdsProperties.Messaging.MessagingServiceConfig config = new CdsProperties.Messaging.MessagingServiceConfig("cfg");
 		config.setBinding("eb-mt-tests-eb");
@@ -29,19 +34,38 @@ void testEmit_TenantNotSupported() {
 		configurer.environmentConfigurations();
 		configurer.serviceConfigurations();
 		configurer.eventHandlerConfigurations();
-		CdsRuntime runtime = configurer.complete();
+		runtime = configurer.complete();
+		svc = runtime.getServiceCatalog().getServices(EventHubMessagingService.class).findFirst().get();
+	}
 
+	@Test
+	void testEmit_TenantNotSupported() {
 		ContextualizedServiceException e = Assertions.assertThrows(ContextualizedServiceException.class, () -> emitMessage(runtime));
 		assertEquals(EventHubErrorStatuses.EVENT_HUB_TENANT_CONTEXT_MISSING, e.getErrorStatus());
 	}
 
 	private void emitMessage(CdsRuntime runtime) {
-		EventHubMessagingService svc = runtime.getServiceCatalog().getServices(EventHubMessagingService.class).findFirst().get();
 		Map<String, Object> data = new HashMap<>();
 		data.put("msg", "my msg 1");
 		Map<String, Object> headers = new HashMap<>();
 		headers.put("msg_header", "my header 1");
-
 		svc.emit("sap.cdscpoc.myobject.myoperation.v1", data, headers);
 	}
+
+	@Test
+	void testResolveSourceSuffixUsesSubaccountId() {
+		Map<String, Object> headers = new HashMap<>();
+		headers.put(EventHubMessagingService.KEY_SUBACCOUNT_ID, "my-subaccount");
+		String suffix = svc.resolveSourceSuffix(headers, "tenant-id");
+		assertEquals("my-subaccount", suffix);
+		assertFalse(headers.containsKey(EventHubMessagingService.KEY_SUBACCOUNT_ID));
+	}
+
+	@Test
+	void testResolveSourceSuffixFallsBackToTenant() {
+		Map<String, Object> headers = new HashMap<>();
+		String suffix = svc.resolveSourceSuffix(headers, "tenant-id");
+		assertEquals("tenant-id", suffix);
+		assertNull(headers.get(EventHubMessagingService.KEY_SUBACCOUNT_ID));
+	}
 }