GH-685 - Introduce spring.modulith.events.jdbc.schema to allow defining the database schema for the event_publication table.

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializer.java

@@ -15,7 +15,15 @@
  */
 package org.springframework.modulith.events.jdbc;
 
+import java.sql.Connection;
+
+import javax.sql.DataSource;
+
 import org.springframework.beans.factory.InitializingBean;
+import org.springframework.core.io.ResourceLoader;
+import org.springframework.jdbc.core.JdbcTemplate;
+import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
+import org.springframework.util.ObjectUtils;
 
 /**
  * Initializes the DB schema used to store events
@@ -23,5 +31,62 @@ import org.springframework.beans.factory.InitializingBean;
  * @author Dmitry Belyaev
  * @author Björn Kieling
  * @author Oliver Drotbohm
+ * @author Raed Ben Hamouda
  */
-interface DatabaseSchemaInitializer extends InitializingBean {}
+class DatabaseSchemaInitializer implements InitializingBean {
+
+	private final DataSource dataSource;
+	private final ResourceLoader resourceLoader;
+	private final DatabaseType databaseType;
+	private final JdbcTemplate jdbcTemplate;
+	private final JdbcConfigurationProperties properties;
+
+	DatabaseSchemaInitializer(DataSource dataSource, ResourceLoader resourceLoader, DatabaseType databaseType,
+			JdbcTemplate jdbcTemplate, JdbcConfigurationProperties properties) {
+
+		this.dataSource = dataSource;
+		this.resourceLoader = resourceLoader;
+		this.databaseType = databaseType;
+		this.jdbcTemplate = jdbcTemplate;
+		this.properties = properties;
+	}
+
+	@Override
+	public void afterPropertiesSet() throws Exception {
+
+		String initialSchema;
+		try (Connection connection = dataSource.getConnection()) {
+
+			initialSchema = connection.getSchema();
+
+			String schemaName = properties.getSchema();
+			if (!ObjectUtils.isEmpty(schemaName)) { // A schema name has been specified.
+
+				if (eventPublicationTableExists(jdbcTemplate, schemaName)) {
+					return;
+				}
+
+				jdbcTemplate.execute("CREATE SCHEMA IF NOT EXISTS " + schemaName);
+				jdbcTemplate.execute(databaseType.sqlStatementSetSchema(schemaName));
+			}
+
+			var locator = new DatabaseSchemaLocator(resourceLoader);
+			new ResourceDatabasePopulator(locator.getSchemaResource(databaseType)).execute(dataSource);
+
+			// Return to the initial schema.
+			if (initialSchema != null) {
+				jdbcTemplate.execute(databaseType.sqlStatementSetSchema(initialSchema));
+			}
+		}
+	}
+
+	private boolean eventPublicationTableExists(JdbcTemplate jdbcTemplate, String schema) {
+		String query = """
+				SELECT COUNT(*)
+				FROM information_schema.tables
+				WHERE table_schema = ? AND table_name = 'EVENT_PUBLICATION'
+				""";
+		Integer count = jdbcTemplate.queryForObject(query, Integer.class, schema);
+		return count != null && count > 0;
+	}
+}

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseType.java

@@ -24,6 +24,7 @@ import org.springframework.util.Assert;
  * @author Dmitry Belyaev
  * @author Björn Kieling
  * @author Oliver Drotbohm
+ * @author Raed Ben Hamouda
  */
 enum DatabaseType {
 
@@ -75,4 +76,11 @@ enum DatabaseType {
 	String getSchemaResourceFilename() {
 		return "/schema-" + value + ".sql";
 	}
+
+	String sqlStatementSetSchema(String schema) {
+		return switch (this) {
+			case MYSQL, H2, HSQLDB -> "SET SCHEMA " + schema;
+			case POSTGRES -> "SET search_path TO " + schema;
+		};
+	}
 }

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcConfigurationProperties.java

@@ -0,0 +1,62 @@
+/*
+ * Copyright 2022-2024 the original author or authors.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *      https://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.modulith.events.jdbc;
+
+import org.springframework.boot.context.properties.ConfigurationProperties;
+import org.springframework.boot.context.properties.bind.ConstructorBinding;
+import org.springframework.boot.context.properties.bind.DefaultValue;
+import org.springframework.lang.Nullable;
+
+/**
+ * Configuration properties for JDBC.
+ *
+ * @author Raed Ben Hamouda
+ */
+@ConfigurationProperties(prefix = "spring.modulith.events.jdbc")
+class JdbcConfigurationProperties {
+
+	private final boolean schemaInitializationEnabled;
+	private final String schema;
+
+	/**
+	 * Creates a new {@link JdbcConfigurationProperties} instance.
+	 *
+	 * @param schemaInitializationEnabled whether to initialize the JDBC event publication schema. Defaults to {@literal false}.
+	 * @param schema the schema name of event publication table, can be {@literal null}.
+	 */
+	@ConstructorBinding
+	JdbcConfigurationProperties(@DefaultValue("false") boolean schemaInitializationEnabled, @Nullable String schema) {
+
+		this.schemaInitializationEnabled = schemaInitializationEnabled;
+		this.schema = schema;
+	}
+
+	/**
+	 * Whether to initialize the JDBC event publication schema.
+	 */
+	public boolean isSchemaInitializationEnabled() {
+		return schemaInitializationEnabled;
+	}
+
+	/**
+	 * The name of the schema where the event publication table resides.
+	 *
+	 * @return can be {@literal null}.
+	 */
+	public String getSchema() {
+		return schema;
+	}
+}

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationAutoConfiguration.java

@@ -21,11 +21,11 @@ import javax.sql.DataSource;
 
 import org.springframework.boot.autoconfigure.AutoConfigureBefore;
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
+import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.io.ResourceLoader;
 import org.springframework.jdbc.core.JdbcTemplate;
-import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator;
 import org.springframework.jdbc.support.JdbcUtils;
 import org.springframework.modulith.events.config.EventPublicationAutoConfiguration;
 import org.springframework.modulith.events.config.EventPublicationConfigurationExtension;
@@ -35,9 +35,11 @@ import org.springframework.modulith.events.core.EventSerializer;
  * @author Dmitry Belyaev
  * @author Björn Kieling
  * @author Oliver Drotbohm
+ * @author Raed Ben Hamouda
  */
 @Configuration(proxyBeanMethods = false)
 @AutoConfigureBefore(EventPublicationAutoConfiguration.class)
+@EnableConfigurationProperties(JdbcConfigurationProperties.class)
 class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigurationExtension {
 
 	@Bean
@@ -47,22 +49,17 @@ class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigura
 
 	@Bean
 	JdbcEventPublicationRepository jdbcEventPublicationRepository(JdbcTemplate jdbcTemplate,
-			EventSerializer serializer, DatabaseType databaseType) {
+			EventSerializer serializer, DatabaseType databaseType, JdbcConfigurationProperties properties) {
 
-		return new JdbcEventPublicationRepository(jdbcTemplate, serializer, databaseType);
+		return new JdbcEventPublicationRepository(jdbcTemplate, serializer, databaseType, properties);
 	}
 
 	@Bean
 	@ConditionalOnProperty(name = "spring.modulith.events.jdbc.schema-initialization.enabled", havingValue = "true")
 	DatabaseSchemaInitializer databaseSchemaInitializer(DataSource dataSource, ResourceLoader resourceLoader,
-			DatabaseType databaseType) {
+			DatabaseType databaseType, JdbcTemplate jdbcTemplate, JdbcConfigurationProperties properties) {
 
-		return () -> {
-
-			var locator = new DatabaseSchemaLocator(resourceLoader);
-
-			new ResourceDatabasePopulator(locator.getSchemaResource(databaseType)).execute(dataSource);
-		};
+		return new DatabaseSchemaInitializer(dataSource, resourceLoader, databaseType, jdbcTemplate, properties);
 	}
 
 	private static String fromDataSource(DataSource dataSource) {

spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java

@@ -40,6 +40,7 @@ import org.springframework.modulith.events.core.TargetEventPublication;
 import org.springframework.transaction.annotation.Transactional;
 import org.springframework.util.Assert;
 import org.springframework.util.ClassUtils;
+import org.springframework.util.ObjectUtils;
 
 /**
  * JDBC-based repository to store {@link TargetEventPublication}s.
@@ -47,78 +48,12 @@ import org.springframework.util.ClassUtils;
  * @author Dmitry Belyaev
  * @author Björn Kieling
  * @author Oliver Drotbohm
+ * @author Raed Ben Hamouda
  */
 class JdbcEventPublicationRepository implements EventPublicationRepository, BeanClassLoaderAware {
 
 	private static final Logger LOGGER = LoggerFactory.getLogger(JdbcEventPublicationRepository.class);
 
-	private static final String SQL_STATEMENT_INSERT = """
-			INSERT INTO EVENT_PUBLICATION (ID, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT)
-			VALUES (?, ?, ?, ?, ?)
-			""";
-
-	private static final String SQL_STATEMENT_FIND_COMPLETED = """
-			SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT
-			FROM EVENT_PUBLICATION
-			WHERE COMPLETION_DATE IS NOT NULL
-			ORDER BY PUBLICATION_DATE ASC
-			""";
-
-	private static final String SQL_STATEMENT_FIND_UNCOMPLETED = """
-			SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT
-			FROM EVENT_PUBLICATION
-			WHERE COMPLETION_DATE IS NULL
-			ORDER BY PUBLICATION_DATE ASC
-			""";
-
-	private static final String SQL_STATEMENT_FIND_UNCOMPLETED_BEFORE = """
-			SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT
-			FROM EVENT_PUBLICATION
-			WHERE
-					COMPLETION_DATE IS NULL
-					AND PUBLICATION_DATE < ?
-			ORDER BY PUBLICATION_DATE ASC
-			""";
-
-	private static final String SQL_STATEMENT_UPDATE_BY_EVENT_AND_LISTENER_ID = """
-			UPDATE EVENT_PUBLICATION
-			SET COMPLETION_DATE = ?
-			WHERE
-					LISTENER_ID = ?
-					AND SERIALIZED_EVENT = ?
-			""";
-
-	private static final String SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID = """
-			SELECT *
-			FROM EVENT_PUBLICATION
-			WHERE
-					SERIALIZED_EVENT = ?
-					AND LISTENER_ID = ?
-					AND COMPLETION_DATE IS NULL
-			ORDER BY PUBLICATION_DATE
-			""";
-
-	private static final String SQL_STATEMENT_DELETE = """
-			DELETE
-			FROM EVENT_PUBLICATION
-			WHERE
-					ID IN
-			""";
-
-	private static final String SQL_STATEMENT_DELETE_UNCOMPLETED = """
-			DELETE
-			FROM EVENT_PUBLICATION
-			WHERE
-					COMPLETION_DATE IS NOT NULL
-			""";
-
-	private static final String SQL_STATEMENT_DELETE_UNCOMPLETED_BEFORE = """
-			DELETE
-			FROM EVENT_PUBLICATION
-			WHERE
-					COMPLETION_DATE < ?
-			""";
-
 	private static final int DELETE_BATCH_SIZE = 100;
 
 	private final JdbcOperations operations;
@@ -126,24 +61,38 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	private final DatabaseType databaseType;
 	private ClassLoader classLoader;
 
+	private String sqlStatementInsert;
+	private String sqlStatementFindCompleted;
+	private String sqlStatementFindUncompleted;
+	private String sqlStatementFindUncompletedBefore;
+	private String sqlStatementUpdateByEventAndListenerId;
+	private String sqlStatementFindByEventAndListenerId;
+	private String sqlStatementDelete;
+	private String sqlStatementDeleteUncompleted;
+	private String sqlStatementDeleteUncompletedBefore;
+
 	/**
-	 * Creates a new {@link JdbcEventPublicationRepository} for the given {@link JdbcOperations}, {@link EventSerializer}
-	 * and {@link DatabaseType}.
+	 * Creates a new {@link JdbcEventPublicationRepository} for the given {@link JdbcOperations}, {@link EventSerializer},
+	 * {@link DatabaseType} and {@link JdbcConfigurationProperties}.
 	 *
 	 * @param operations must not be {@literal null}.
 	 * @param serializer must not be {@literal null}.
 	 * @param databaseType must not be {@literal null}.
+	 * @param properties must not be {@literal null}.
 	 */
 	public JdbcEventPublicationRepository(JdbcOperations operations, EventSerializer serializer,
-			DatabaseType databaseType) {
+			DatabaseType databaseType, JdbcConfigurationProperties properties) {
 
 		Assert.notNull(operations, "JdbcOperations must not be null!");
 		Assert.notNull(serializer, "EventSerializer must not be null!");
 		Assert.notNull(databaseType, "DatabaseType must not be null!");
+		Assert.notNull(properties, "JdbcConfigurationProperties must not be null!");
 
 		this.operations = operations;
 		this.serializer = serializer;
 		this.databaseType = databaseType;
+
+		prepareSqlStatements(properties.getSchema());
 	}
 
 	/*
@@ -166,7 +115,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 		var serializedEvent = serializeEvent(publication.getEvent());
 
 		operations.update( //
-				SQL_STATEMENT_INSERT, //
+				sqlStatementInsert, //
 				uuidToDatabase(publication.getIdentifier()), //
 				publication.getEvent().getClass().getName(), //
 				publication.getTargetIdentifier().getValue(), //
@@ -184,7 +133,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	@Transactional
 	public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) {
 
-		operations.update(SQL_STATEMENT_UPDATE_BY_EVENT_AND_LISTENER_ID, //
+		operations.update(sqlStatementUpdateByEventAndListenerId, //
 				Timestamp.from(completionDate), //
 				identifier.getValue(), //
 				serializer.serialize(event));
@@ -199,7 +148,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	public Optional<TargetEventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
 			Object event, PublicationTargetIdentifier targetIdentifier) {
 
-		var result = operations.query(SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, //
+		var result = operations.query(sqlStatementFindByEventAndListenerId, //
 				this::resultSetToPublications, //
 				serializeEvent(event), //
 				targetIdentifier.getValue());
@@ -214,7 +163,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	@Override
 	public List<TargetEventPublication> findCompletedPublications() {
 
-		var result = operations.query(SQL_STATEMENT_FIND_COMPLETED, this::resultSetToPublications);
+		var result = operations.query(sqlStatementFindCompleted, this::resultSetToPublications);
 
 		return result == null ? Collections.emptyList() : result;
 	}
@@ -223,7 +172,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	@Transactional(readOnly = true)
 	@SuppressWarnings("null")
 	public List<TargetEventPublication> findIncompletePublications() {
-		return operations.query(SQL_STATEMENT_FIND_UNCOMPLETED, this::resultSetToPublications);
+		return operations.query(sqlStatementFindUncompleted, this::resultSetToPublications);
 	}
 
 	/*
@@ -233,7 +182,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	@Override
 	public List<TargetEventPublication> findIncompletePublicationsPublishedBefore(Instant instant) {
 
-		var result = operations.query(SQL_STATEMENT_FIND_UNCOMPLETED_BEFORE,
+		var result = operations.query(sqlStatementFindUncompletedBefore,
 				this::resultSetToPublications, Timestamp.from(instant));
 
 		return result == null ? Collections.emptyList() : result;
@@ -249,7 +198,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 		var dbIdentifiers = identifiers.stream().map(databaseType::uuidToDatabase).toList();
 
 		batch(dbIdentifiers, DELETE_BATCH_SIZE)
-				.forEach(it -> operations.update(SQL_STATEMENT_DELETE.concat(toParameterPlaceholders(it.length)), it));
+				.forEach(it -> operations.update(sqlStatementDelete.concat(toParameterPlaceholders(it.length)), it));
 	}
 
 	/*
@@ -258,7 +207,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 	 */
 	@Override
 	public void deleteCompletedPublications() {
-		operations.execute(SQL_STATEMENT_DELETE_UNCOMPLETED);
+		operations.execute(sqlStatementDeleteUncompleted);
 	}
 
 	/*
@@ -270,7 +219,85 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean
 
 		Assert.notNull(instant, "Instant must not be null!");
 
-		operations.update(SQL_STATEMENT_DELETE_UNCOMPLETED_BEFORE, Timestamp.from(instant));
+		operations.update(sqlStatementDeleteUncompletedBefore, Timestamp.from(instant));
+	}
+
+	/**
+	 * Prepares SQL statements based on the provided schema name.
+	 *
+	 * @param schema can be {@literal null}.
+	 */
+	private void prepareSqlStatements(String schema) {
+
+		String tableName = ObjectUtils.isEmpty(schema) ? "EVENT_PUBLICATION" : schema + ".EVENT_PUBLICATION";
+
+
+		sqlStatementInsert = """
+				INSERT INTO %s (ID, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT)
+				VALUES (?, ?, ?, ?, ?)
+				""".formatted(tableName);
+
+		sqlStatementFindCompleted = """
+				SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT
+				FROM %s
+				WHERE COMPLETION_DATE IS NOT NULL
+				ORDER BY PUBLICATION_DATE ASC
+				""".formatted(tableName);
+
+		sqlStatementFindUncompleted = """
+				SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT
+				FROM %s
+				WHERE COMPLETION_DATE IS NULL
+				ORDER BY PUBLICATION_DATE ASC
+				""".formatted(tableName);
+
+		sqlStatementFindUncompletedBefore = """
+				SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT
+				FROM %s
+				WHERE
+				        COMPLETION_DATE IS NULL
+				        AND PUBLICATION_DATE < ?
+				ORDER BY PUBLICATION_DATE ASC
+				""".formatted(tableName);
+
+		sqlStatementUpdateByEventAndListenerId = """
+				UPDATE %s
+				SET COMPLETION_DATE = ?
+				WHERE
+				        LISTENER_ID = ?
+				        AND SERIALIZED_EVENT = ?
+				""".formatted(tableName);
+
+		sqlStatementFindByEventAndListenerId = """
+				SELECT *
+				FROM %s
+				WHERE
+				        SERIALIZED_EVENT = ?
+				        AND LISTENER_ID = ?
+				        AND COMPLETION_DATE IS NULL
+				ORDER BY PUBLICATION_DATE
+				""".formatted(tableName);
+
+		sqlStatementDelete = """
+				DELETE
+				FROM %s
+				WHERE
+				        ID IN
+				""".formatted(tableName);
+
+		sqlStatementDeleteUncompleted = """
+				DELETE
+				FROM %s
+				WHERE
+				        COMPLETION_DATE IS NOT NULL
+				""".formatted(tableName);
+
+		sqlStatementDeleteUncompletedBefore = """
+				DELETE
+				FROM %s
+				WHERE
+				        COMPLETION_DATE < ?
+				""".formatted(tableName);
 	}
 
 	private String serializeEvent(Object event) {

spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/META-INF/spring-configuration-metadata.json

@@ -1,10 +0,0 @@
-{
-	"properties": [
-		{
-			"name": "spring.modulith.events.jdbc.schema-initialization.enabled",
-			"type": "java.lang.Boolean",
-			"description": "Whether to initialize the JDBC event publication schema.",
-			"defaultValue": "false"
-		}
-	]
-}

spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java

@@ -50,16 +50,16 @@ import org.testcontainers.junit.jupiter.Testcontainers;
  * @author Dmitry Belyaev
  * @author Björn Kieling
  * @author Oliver Drotbohm
+ * @author Raed Ben Hamouda
  */
 class JdbcEventPublicationRepositoryIntegrationTests {
 
 	static final PublicationTargetIdentifier TARGET_IDENTIFIER = PublicationTargetIdentifier.of("listener");
 
-	@JdbcTest(properties = "spring.modulith.events.jdbc.schema-initialization.enabled=true")
 	@Import(TestApplication.class)
 	@Testcontainers(disabledWithoutDocker = true)
 	@ContextConfiguration(classes = JdbcEventPublicationAutoConfiguration.class)
-	abstract class TestBase {
+	static abstract class TestBase {
 
 		@Autowired JdbcOperations operations;
 		@Autowired JdbcEventPublicationRepository repository;
@@ -68,7 +68,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 
 		@BeforeEach
 		void cleanUp() {
-			operations.execute("TRUNCATE TABLE EVENT_PUBLICATION");
+			operations.execute("TRUNCATE TABLE " + table());
 		}
 
 		@Test // GH-3
@@ -205,7 +205,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 			// Store publication
 			repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));
 
-			operations.update("UPDATE EVENT_PUBLICATION SET EVENT_TYPE='abc'");
+			operations.update("UPDATE " + table() + " SET EVENT_TYPE='abc'");
 
 			assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER))
 					.isEmpty();
@@ -230,7 +230,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 			repository.markCompleted(publication, Instant.now());
 			repository.deleteCompletedPublications();
 
-			assertThat(operations.query("SELECT * FROM EVENT_PUBLICATION", (rs, __) -> rs.getString("SERIALIZED_EVENT")))
+			assertThat(operations.query("SELECT * FROM " + table(), (rs, __) -> rs.getString("SERIALIZED_EVENT")))
 					.hasSize(1).element(0).isEqualTo(serializedEvent2);
 		}
 
@@ -256,7 +256,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 			repository.markCompleted(testEvent2, TARGET_IDENTIFIER, now);
 			repository.deleteCompletedPublicationsBefore(now.minusSeconds(15));
 
-			assertThat(operations.query("SELECT * FROM EVENT_PUBLICATION", (rs, __) -> rs.getString("SERIALIZED_EVENT")))
+			assertThat(operations.query("SELECT * FROM " + table(), (rs, __) -> rs.getString("SERIALIZED_EVENT")))
 					.hasSize(1).element(0).isEqualTo(serializedEvent2);
 		}
 
@@ -310,6 +310,8 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 					.isEqualTo(event);
 		}
 
+		abstract String table();
+
 		private TargetEventPublication createPublication(Object event) {
 
 			var token = event.toString();
@@ -321,23 +323,95 @@ class JdbcEventPublicationRepositoryIntegrationTests {
 		}
 	}
 
+	@Nested
+	@JdbcTest(properties = "spring.modulith.events.jdbc.schema-initialization.enabled=true")
+	static class WithNoDefinedSchemaName extends TestBase {
+
+		@Override
+		String table() {
+			return "EVENT_PUBLICATION";
+		}
+	}
+
+	@Nested
+	@JdbcTest(properties = {"spring.modulith.events.jdbc.schema-initialization.enabled=true",
+			"spring.modulith.events.jdbc.schema=test"})
+	static class WithDefinedSchemaName extends TestBase {
+
+		@Override
+		String table() {
+			return "test.EVENT_PUBLICATION";
+		}
+	}
+
+	@Nested
+	@JdbcTest(properties = {"spring.modulith.events.jdbc.schema-initialization.enabled=true",
+			"spring.modulith.events.jdbc.schema="})
+	static class WithEmptySchemaName extends TestBase {
+
+		@Override
+		String table() {
+			return "EVENT_PUBLICATION";
+		}
+	}
+
+	// HSQL
 	@Nested
 	@ActiveProfiles("hsqldb")
 	@Testcontainers(disabledWithoutDocker = false)
-	class HSQL extends TestBase {}
+	class HsqlWithNoDefinedSchemaName extends WithNoDefinedSchemaName {}
+
+	@Nested
+	@ActiveProfiles("hsqldb")
+	@Testcontainers(disabledWithoutDocker = false)
+	class HsqlWithDefinedSchemaName extends WithDefinedSchemaName {}
+
+	@Nested
+	@ActiveProfiles("hsqldb")
+	@Testcontainers(disabledWithoutDocker = false)
+	class HsqlWithEmptySchemaName extends WithEmptySchemaName {}
+
+	// H2
+	@Nested
+	@ActiveProfiles("h2")
+	@Testcontainers(disabledWithoutDocker = false)
+	class H2WithNoDefinedSchemaName extends WithNoDefinedSchemaName {}
 
 	@Nested
 	@ActiveProfiles("h2")
 	@Testcontainers(disabledWithoutDocker = false)
-	class H2 extends TestBase {}
+	class H2WithDefinedSchemaName extends WithDefinedSchemaName {}
+
+	@Nested
+	@ActiveProfiles("h2")
+	@Testcontainers(disabledWithoutDocker = false)
+	class H2WithEmptySchemaName extends WithEmptySchemaName {}
+
+	// Postgres
+	@Nested
+	@ActiveProfiles("postgres")
+	class PostgresWithNoDefinedSchemaName extends WithNoDefinedSchemaName {}
 
 	@Nested
 	@ActiveProfiles("postgres")
-	class Postgres extends TestBase {}
+	class PostgresWithDefinedSchemaName extends WithDefinedSchemaName {}
+
+	@Nested
+	@ActiveProfiles("postgres")
+	class PostgresWithEmptySchemaName extends WithEmptySchemaName {}
+
+	// MySQL
+	@Nested
+	@ActiveProfiles("mysql")
+	class MysqlWithNoDefinedSchemaName extends WithNoDefinedSchemaName {}
 
 	@Nested
 	@ActiveProfiles("mysql")
-	class MySQL extends TestBase {}
+	class MysqlWithDefinedSchemaName extends WithDefinedSchemaName {}
+
+	@Nested
+	@ActiveProfiles("mysql")
+	class MysqlWithEmptySchemaName extends WithEmptySchemaName {}
 
 	@Value
 	private static final class TestEvent {

src/docs/antora/modules/ROOT/pages/appendix.adoc

@@ -26,6 +26,10 @@ Can either be the class name of a custom implementation of `ApplicationModuleDet
 |`false`
 |Whether to initialize the JDBC event publication schema.
 
+|`spring.modulith.events.jdbc.schema`
+|
+|The schema name for the event publication table. If not specified, the table will not be schema-qualified.
+
 |`spring.modulith.events.kafka.enable-json`
 |`true`
 |Whether to enable JSON support for `KafkaTemplate`.