The following steps need to be taken in order to enable data-services to support a read replica:
Disable springs open session in view feature
Configure spring property
spring.jpa.open-in-view=false
This feature, which is enabled by default, causes spring to hold onto a once obtained jdbc connection until the request processing finished. This prevents request processing to make use of a read replica if the request needs a mix of read and write transactions to be processed. This is for example the case for all json-rpc requests to data-services.
Configure datasources for the read replica
Data-services mainly uses the datasources csDataSource for the content store and dsDataSource for the document store and everything else. There need to be 2 additional datasources configured corresponding to these that make use of the read replica.
@Configuration
protected static class ReadDataSourceConfiguration {
@ConfigurationProperties("spring.datasources.dataservicesread")
@Bean
public DataSourceProperties dsReadDatasourceProperties() {
return new DataSourceProperties();
}
@Bean
public DataSource dsReadDataSource(
@Qualifier("dsReadDatasourceProperties") DataSourceProperties dsReadDatasourceProperties) {
HikariDataSource dataSource = dsReadDatasourceProperties.initializeDataSourceBuilder()
.type(HikariDataSource.class).build();
if (StringUtils.isNotBlank(dsReadDatasourceProperties.getName())) {
dataSource.setPoolName(dsReadDatasourceProperties.getName());
}
return dataSource;
}
@ConfigurationProperties("spring.datasources.contentstoreread")
@Bean
public DataSourceProperties csReadDatasourceProperties() {
return new DataSourceProperties();
}
@Bean
public DataSource csReadDataSource(
@Qualifier("csReadDatasourceProperties") DataSourceProperties csReadDatasourceProperties) {
HikariDataSource dataSource = csReadDatasourceProperties.initializeDataSourceBuilder()
.type(HikariDataSource.class).build();
if (StringUtils.isNotBlank(csReadDatasourceProperties.getName())) {
dataSource.setPoolName(csReadDatasourceProperties.getName());
}
return dataSource;
}
}
Those could then be configured via:
spring.datasources.dataservicesread.username=data-services-ds-user-read
spring.datasources.dataservicesread.password=*****
spring.datasources.dataservicesread.url=jdbc:postgresql://localhost:5432/data-services-ds-read
spring.datasources.contentstoreread.username=data-services-cs-user-read
spring.datasources.contentstoreread.password=*****
spring.datasources.contentstoreread.url=jdbc:postgresql://localhost:5434/data-services-cs-read
Choose a datasource at runtime depending on the transaction attributes
The idea is to choose whether to use dsDataSource or dsReadDataSource based on whether
@Transaction(readOnly = false) or @Transactional(readOnly = true) is specified. Spring provides org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource for such purposes.
However the tricky part is that at the time the transaction manager of spring requests a jdbc connection from the datasource the transaction is not started yet. So there has to be a workaround to determine what the transaction will be that is going to be started. To solve that we can make use of a org.springframework.transaction.TransactionExecutionListener that is going to signal the AbstractRoutingDataSource implementation via application events what the transaction will be that is currently being started.
And another tricky part is the initialization order. Liquibase and Quartz initialize database stuff before springs transaction manager is initialized. So we need to account for this to avoid creating a cyclic bean dependency during application startup by using a javax.inject.Provider that will provide a transaction manager once it has been initialized in the application context. Otherwise we default to the writeable datasource in order to allow Liquibase and Quartz to do database initializations.
And the last tricky part is that datasources defined by data-services can not be overriden easily because for example dsDataSource is marked as @Primary. Therefore we use a org.springframework.beans.factory.config.BeanPostProcessor to alter those beans after they were initialized according to our needs.
Then the actual AbstractRoutingDataSource implementation looks like this:
import org.springframework.context.ApplicationListener;
import org.springframework.core.NamedThreadLocal;
import org.springframework.jdbc.datasource.lookup.AbstractRoutingDataSource;
import org.springframework.transaction.PlatformTransactionManager;
import com.mgmtp.project.jpasupport.transaction.AfterTransactionCompletedEvent;
import com.mgmtp.project.jpasupport.transaction.BeforeTransactionStartedEvent;
import com.mgmtp.project.jpasupport.transaction.TransactionLifecycleEvent;
import jakarta.inject.Provider;
import lombok.RequiredArgsConstructor;
@RequiredArgsConstructor
public class ReadWriteAwareRoutingDataSource extends AbstractRoutingDataSource
implements ApplicationListener<TransactionLifecycleEvent> {
public static enum DataSourceKey {
READ, WRITE, UNDEFINED;
}
private final ThreadLocal<DataSourceKey> dataSourceKeyThreadLocal = NamedThreadLocal
.withInitial("dataSourceKeyThreadLocal", () -> DataSourceKey.UNDEFINED);
private final Provider<PlatformTransactionManager> platformTransactionManagerProvider;
@Override
protected Object determineCurrentLookupKey() {
DataSourceKey dataSourceKey = dataSourceKeyThreadLocal.get();
if (dataSourceKey == DataSourceKey.UNDEFINED) {
dataSourceKeyThreadLocal.remove();
return DataSourceKey.WRITE;
} else {
return dataSourceKey;
}
}
@Override
public void onApplicationEvent(TransactionLifecycleEvent event) {
if (event instanceof BeforeTransactionStartedEvent beforeStarted) {
if (platformTransactionManagerProvider.get().equals(beforeStarted.getPlatformTransactionManager())) {
if (beforeStarted.getTransaction().isReadOnly()) {
dataSourceKeyThreadLocal.set(DataSourceKey.READ);
} else {
dataSourceKeyThreadLocal.set(DataSourceKey.WRITE);
}
}
}
if (event instanceof AfterTransactionCompletedEvent afterCompleted) {
if (platformTransactionManagerProvider.get().equals(afterCompleted.getPlatformTransactionManager())) {
dataSourceKeyThreadLocal.remove();
}
}
}
}
The TransactionExecutionListener implementation for notifying the data source implementation above is:
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.TransactionExecution;
import org.springframework.transaction.TransactionExecutionListener;
import lombok.RequiredArgsConstructor;
@RequiredArgsConstructor
public class TransactionStateNotifyingTransactionExecutionListener implements TransactionExecutionListener {
private final ApplicationEventPublisher applicationEventPublisher;
private final PlatformTransactionManager platformTransactionManager;
@Override
public void beforeBegin(TransactionExecution transaction) {
applicationEventPublisher
.publishEvent(new BeforeTransactionStartedEvent(platformTransactionManager, transaction));
}
public void afterCommit(TransactionExecution transaction, Throwable commitFailure) {
applicationEventPublisher
.publishEvent(new AfterTransactionCompletedEvent(platformTransactionManager, transaction));
}
public void afterRollback(TransactionExecution transaction, Throwable rollbackFailure) {
applicationEventPublisher
.publishEvent(new AfterTransactionCompletedEvent(platformTransactionManager, transaction));
}
}
Then we can wire this all up with the following spring configuration:
import java.util.Map;
import java.util.concurrent.ConcurrentHashMap;
import javax.sql.DataSource;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.BeanFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.BeanPostProcessor;
import org.springframework.boot.autoconfigure.jdbc.DataSourceProperties;
import org.springframework.boot.context.properties.ConfigurationProperties;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.ConfigurableApplicationContext;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.transaction.ConfigurableTransactionManager;
import org.springframework.transaction.PlatformTransactionManager;
import com.mgmtp.project.jpasupport.datasource.ReadWriteAwareRoutingDataSource;
import com.mgmtp.project.jpasupport.datasource.ReadWriteAwareRoutingDataSource.DataSourceKey;
import com.mgmtp.project.jpasupport.transaction.TransactionStateNotifyingTransactionExecutionListener;
import com.zaxxer.hikari.HikariDataSource;
import jakarta.inject.Provider;
@Configuration
public class ReadWriteDatasourceConfiguration {
private DataSource createReadWriteDataSource(Provider<PlatformTransactionManager> platformTransactionManager,
DataSource writeDataSource, DataSource readDataSource, ConfigurableApplicationContext applicationContext) {
ReadWriteAwareRoutingDataSource readWriteAwareRoutingDataSource = new ReadWriteAwareRoutingDataSource(
platformTransactionManager);
readWriteAwareRoutingDataSource
.setTargetDataSources(Map.of(DataSourceKey.READ, readDataSource, DataSourceKey.WRITE, writeDataSource));
readWriteAwareRoutingDataSource.setDefaultTargetDataSource(writeDataSource);
readWriteAwareRoutingDataSource.initialize();
applicationContext.addApplicationListener(readWriteAwareRoutingDataSource);
return readWriteAwareRoutingDataSource;
}
@Bean
public BeanPostProcessor decorateWriteDataSourcesWithReadWriteBeanPostProcessor(BeanFactory beanFactory,
ApplicationEventPublisher applicationEventPublisher, ConfigurableApplicationContext applicationContext) {
return new BeanPostProcessor() {
private final Map<String, String> dataSourcesToDecorate = Map.of("dsDataSource", "dsTransactionManager",
"csDataSource", "csTransactionManager");
private final Map<String, String> dataSourceReadDataSource = Map.of("dsDataSource", "dsReadDataSource",
"csDataSource", "csReadDataSource");
private final Map<String, Boolean> transactionManagerReadyState = new ConcurrentHashMap<String, Boolean>(
Map.of("dsTransactionManager", false, "csTransactionManager", false));
@Override
public Object postProcessAfterInitialization(Object bean, String beanName) throws BeansException {
if (bean instanceof PlatformTransactionManager platformTransactionManager
&& bean instanceof ConfigurableTransactionManager configurableTransactionManager
&& transactionManagerReadyState.containsKey(beanName)) {
transactionManagerReadyState.put(beanName, true);
configurableTransactionManager.addListener(
new TransactionStateNotifyingTransactionExecutionListener(applicationEventPublisher,
platformTransactionManager));
}
if (bean instanceof DataSource dataSource && dataSourcesToDecorate.containsKey(beanName)) {
String transactionManagerBeanName = dataSourcesToDecorate.get(beanName);
Provider<PlatformTransactionManager> platformTransactionManagerProvider = new ProvideBeanWhenReadyProvider<>(
transactionManagerBeanName, PlatformTransactionManager.class,
this::isTransactionManagerReady, beanFactory);
DataSource readDataSource = beanFactory.getBean(dataSourceReadDataSource.get(beanName),
DataSource.class);
return createReadWriteDataSource(platformTransactionManagerProvider, dataSource, readDataSource,
applicationContext);
} else {
return bean;
}
}
private boolean isTransactionManagerReady(String transactionManagerBeanName) {
return Boolean.TRUE.equals(transactionManagerReadyState.get(transactionManagerBeanName));
}
};
}
}
Add a read-only json rpc endpoint
The json rpc endpoint provided by data-services will always execute write transactions despite the RPC operations being annotated with @Transaction(writeOnly = true) for those that actually only do a read operation. This is because of com.mgmtp.a12.dataservices.rpc.internal.JsonRpcOperationDispatcher.handleRequest(InputStream, OutputStream) is being annotated with @Transactional.
The original json rpc controller is left untouched. However the read-only one will be available under /api/v2/rpc-read and look like this:
import java.io.IOException;
import java.io.InputStream;
import org.springframework.core.io.ByteArrayResource;
import org.springframework.http.MediaType;
import org.springframework.http.ResponseEntity;
import org.springframework.web.bind.annotation.PostMapping;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RestController;
import com.mgmtp.a12.dataservices.server.rpc.JsonRpcControllerImpl;
import com.mgmtp.a12.dataservices.server.uaa.SecuredController;
import lombok.RequiredArgsConstructor;
@RequestMapping("#{@dataServicesCoreProperties.server.contextPath}/v2/rpc-read")
@SecuredController
@RestController
@RequiredArgsConstructor
public class ReadOnlyJsonRpcController extends JsonRpcControllerImpl {
private final ReadWriteAwareJsonRpcOperationDispatcher readWriteAwareJsonRpcOperationDispatcher;
@Override
@PostMapping(consumes = { MediaType.APPLICATION_JSON_VALUE }, produces = { MediaType.APPLICATION_JSON_VALUE })
public ResponseEntity<ByteArrayResource> jsonRpc(InputStream request, String requestId) throws IOException {
return readWriteAwareJsonRpcOperationDispatcher.runAsReadOnlyRpcOperation(() -> super.jsonRpc(request, requestId));
}
}
So we are replacing the original JsonRpcOperationDispatcher with this one:
import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.Set;
import java.util.concurrent.Callable;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.core.NamedThreadLocal;
import org.springframework.transaction.annotation.Propagation;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.transaction.support.TransactionTemplate;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mgmtp.a12.dataservices.configuration.DataServicesCoreProperties;
import com.mgmtp.a12.dataservices.relationship.internal.RelationshipLinkValidationListener;
import com.mgmtp.a12.dataservices.rpc.internal.JsonRpcOperationDispatcher;
import com.mgmtp.a12.dataservices.utils.internal.JsonUtils;
public class ReadWriteAwareJsonRpcOperationDispatcher extends JsonRpcOperationDispatcher {
@SuppressWarnings("serial")
private static final class ReadOnlyTransactionRequestHandlingException extends RuntimeException {
private ReadOnlyTransactionRequestHandlingException(IOException cause) {
super(cause);
}
@Override
public IOException getCause() {
return (IOException) super.getCause();
}
}
public interface IOExceptionThrowing<T> extends Callable<T> {
@Override
T call() throws IOException;
}
private final NamedThreadLocal<Boolean> readOnlyRpcOperationHolder = new NamedThreadLocal<Boolean>("readOnlyRpcOperation");
private final TransactionTemplate readOnlyTransactionTemplate;
private final TransactionTemplate writeTransactionTemplate;
public ReadWriteAwareJsonRpcOperationDispatcher(Set<String> allowedOperations,
RelationshipLinkValidationListener linkValidator, ApplicationEventPublisher applicationEventPublisher,
ObjectMapper objectMapper, DataServicesCoreProperties dataServicesCoreProperties, boolean spelAllowed,
JsonUtils jsonUtils, boolean debugRpcResponses, TransactionTemplate readOnlyTransactionTemplate, TransactionTemplate writeTransactionTemplate) {
super(allowedOperations, linkValidator, applicationEventPublisher, objectMapper, dataServicesCoreProperties,
spelAllowed, jsonUtils, debugRpcResponses);
this.readOnlyTransactionTemplate = readOnlyTransactionTemplate;
this.writeTransactionTemplate = writeTransactionTemplate;
}
public <T> T runAsReadOnlyRpcOperation(IOExceptionThrowing<T> action) throws IOException {
readOnlyRpcOperationHolder.set(true);
try {
return action.call();
} finally {
readOnlyRpcOperationHolder.remove();
}
}
@Transactional(propagation = Propagation.NEVER)
@Override
public int handleRequest(InputStream input, OutputStream output) throws IOException {
if(Boolean.TRUE.equals(readOnlyRpcOperationHolder.get())) {
try {
return readOnlyTransactionTemplate.<Integer>execute((status) -> {
try {
return super.handleRequest(input, output);
} catch (IOException e) {
throw new ReadOnlyTransactionRequestHandlingException(e);
}
});
} catch (ReadOnlyTransactionRequestHandlingException e) {
throw e.getCause();
}
} else {
try {
return writeTransactionTemplate.<Integer>execute((status) -> {
try {
return super.handleRequest(input, output);
} catch (IOException e) {
throw new ReadOnlyTransactionRequestHandlingException(e);
}
});
} catch (ReadOnlyTransactionRequestHandlingException e) {
throw e.getCause();
}
}
}
}
which use a ThreadLocal to know whether it should start a writeable transaction or a readable one.
Then we only need to adopt some copy-pasted spring configuration from data-services to use that dispatcher instead:
import java.util.Collections;
import java.util.Optional;
import java.util.Set;
import org.springframework.context.ApplicationEventPublisher;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Primary;
import org.springframework.transaction.PlatformTransactionManager;
import org.springframework.transaction.support.DefaultTransactionDefinition;
import org.springframework.transaction.support.TransactionTemplate;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mgmtp.a12.dataservices.configuration.DataServicesCoreProperties;
import com.mgmtp.a12.dataservices.relationship.internal.RelationshipLinkValidationListener;
import com.mgmtp.a12.dataservices.utils.internal.JsonUtils;
import com.mgmtp.project.server.controller.ReadWriteAwareJsonRpcOperationDispatcher;
import lombok.RequiredArgsConstructor;
@Configuration
@RequiredArgsConstructor
public class ReadOnlyJsonRpcConfiguration {
@Bean
@Primary
public ReadWriteAwareJsonRpcOperationDispatcher readWriteAwareJsonRpcOperationDispatcher(
RelationshipLinkValidationListener linkValidator,
ApplicationEventPublisher applicationEventPublisher,
ObjectMapper objectMapper,
DataServicesCoreProperties dataServicesCoreProperties,
JsonUtils jsonUtils,
PlatformTransactionManager platformTransactionManager) {
DefaultTransactionDefinition readOnlyTransactionDefinition = new DefaultTransactionDefinition();
readOnlyTransactionDefinition.setReadOnly(true);
DefaultTransactionDefinition writeTransactionDefinition = new DefaultTransactionDefinition();
return new ReadWriteAwareJsonRpcOperationDispatcher(getAllowedOperations(dataServicesCoreProperties),
linkValidator,
applicationEventPublisher,
objectMapper,
dataServicesCoreProperties,
isSpelAllowed(dataServicesCoreProperties),
jsonUtils,
false,
new TransactionTemplate(platformTransactionManager, readOnlyTransactionDefinition),
new TransactionTemplate(platformTransactionManager, writeTransactionDefinition));
}
private boolean isSpelAllowed(DataServicesCoreProperties dataServicesCoreProperties) {
return Optional.ofNullable(dataServicesCoreProperties)
.map(DataServicesCoreProperties::getJsonRpc)
.map(DataServicesCoreProperties.JsonRpc::getSpel)
.map(DataServicesCoreProperties.JsonRpc.Spel::isEnabled)
.orElse(false);
}
private Set<String> getAllowedOperations(DataServicesCoreProperties dataServicesCoreProperties) {
return Optional.ofNullable(dataServicesCoreProperties)
.map(DataServicesCoreProperties::getJsonRpc)
.map(DataServicesCoreProperties.JsonRpc::getAllowedOperations)
.orElse(Collections.emptySet());
}
}
A better approach for the json-rpc endpoint would be to look at the rpc operations and whether they are all read-only or not. But this execise is left to the proficient reader.