Kristina Kraljevic

Intro While a single data source can be sufficient for many web applications, nowadays, relying solely on one data source is not the case for most complex or large-scale applications. Since we live in the era of big data, the common reason for this change that comes to mind is the demand for data itself. Beyond the obvious, multiple data sources could be used for other important aspects, such as security and performance.  Another example would be systems with high throughput, especially with heavy read traffic -  performance improvement could be accomplished by using a read-only instance or replica of the primary data source to handle all reading requests, leaving more resources available for executing other modifying actions over the primary data source. Spring Boot almost seamlessly handles one data source in terms of connectivity and transactions with an auto-configured connection pool and transaction manager. Auto-configuration doesn't apply to additional data sources, especially when handling transactions. Let’s take a look into that! Single Data Source configuration As mentioned above, for a single data source, Spring Boot provides auto-configuration that covers the basic setup for the connection pool. Connection  pool  If not specified otherwise, it is managed by HikariCP and included into Spring Data JPA and Spring Data JDBC starter dependencies with the default configuration: maximumPoolSize - 10 connectionTimeout - 30s idleTimeout - 10min maxLifetime - 30min autoCommit - true The rest of the connection pool properties can be found here. Transaction manager  Managed by DataSourceTransactionManager that will implicitly take care of transactions with default setup: a connection to data source is bounded to the current thread (thread-local strategy) ensuring the usage of the same connection during the transaction lifecycle supports nested transactions by default* default propagation is REQUIRED default isolation is DEFAULT (depends on the used data store) When propagation is set to REQUIRED, Spring will ensure that triggered transactional action operates within the transactional context. If that context already exists (active transaction), the actions will be executed as part of it. Otherwise, Spring will create a new one. When the isolation level is set to DEFAULT, it will inherit the isolation level of the data store in use. For example, READ_COMMITTED is the default isolation level for Postgres database that prevents dirty reads - reading uncommitted changes. The properties for data source have the prefix “spring.datasource” and to create a single data source, you should provide the url along with a username and password if needed. Even the type of driver could be recognized by Spring from the url. Often, the connection pool configuration highly depends on your application and resources available on the data source side that could be configured too. spring.datasource.url=jdbc:postgresql://localhost:5432/postgres spring.datasource.username=db_user spring.datasource.password=PassWord123 spring.datasource.hikari.maximumPoolSize=30 spring.datasource.hikari.connectionTimeout=60000 spring.datasource.hikari.maxLifetime=30000 The list of available configuration properties can be found here. After adding the data source configuration properties, we are ready to set up the data access layer and run some queries quickly.   Multiple Data Sources in Spring  There are two main approaches to configuring multiple data sources in Spring Boot: using auto-configuration for one data source and custom configuration for additional sources or manually configuring all data sources for consistency and control. Manually configuring all data sources ensures a consistent setup and reduces the risk of errors. This approach simplifies debugging and maintenance by avoiding potential conflicts between auto-configured and custom-configured data sources. Let's start with configuration for two data sources, primary and secondary, with configuration prefixes "spring.datasource.primary" and "spring.datasource.secondary" respectively. Spring won't recognize these two configurations, so we need to define them explicitly.  Note that when there are multiple data sources, there should be one annotated as @Primary.  spring.datasource.primary.jdbcUrl=jdbc:postgresql://localhost:5432/postgres spring.datasource.primary.username=db_user_1 spring.datasource.primary.password=PassWord123 spring.datasource.primary.driverClassName=org.postgresql.Driver spring.datasource.primary.maximumPoolSize=40 spring.datasource.primary.minimumIdle=20 spring.datasource.primary.maxLifetime=240000 spring.datasource.primary.idleTimeout=180000 spring.datasource.secondary.jdbcUrl=jdbc:postgresql://localhost:5433/postgres spring.datasource.secondary.username=db_user_2 spring.datasource.secondary.password=OtherPassWord123 spring.datasource.secondary.driverClassName=org.postgresql.Driver spring.datasource.secondary.maximumPoolSize=40 spring.datasource.secondary.minimumIdle=20 spring.datasource.secondary.maxLifetime=240000 spring.datasource.secondary.idleTimeout=180000 For covering both data source and connection pool-related properties HikariConfig could be used for initialization. Picking up these two configurations by Spring is accomplished by defining configuration classes as follows: @Configuration @ConfigurationProperties("spring.datasource.primary") public class DataSourcePrimaryConfig extends HikariConfig { public static final String PRIMARY_DATASOURCE = "primaryDataSource"; @Bean(PRIMARY_DATASOURCE) @Primary public HikariDataSource primaryDataSource() { return new HikariDataSource(this); } } @Configuration @ConfigurationProperties("spring.datasource.secondary") public class DataSourceSecondaryConfig extends HikariConfig { public static final String SECONDARY_DATASOURCE = "secondaryDataSource"; @Bean(SECONDARY_DATASOURCE) public HikariDataSource secondaryDataSource() { return new HikariDataSource(this); } } For this step, HikariDataSource class is used instead of the regular DataSource class for one practical reason - connection pool management is covered by it. This class extends HikariConfig and implements both DataSource and Closeable, which covers all necessary connection pool functionalities and responsibilities.  By invoking a constructor with HikariConfig as a parameter, like in our case, the connection pool is started based on the provided configuration right away. On the other hand, by invoking the default constructor, the connection pool would start when the getConnection method is invoked. Data source bean names are defined as public static variables for further reference when setting up a data access layer - mostly used is JPA, and it will be used for this blog. Let’s proceed to the JPA setup. For both classes from above, entity and transaction manager will be added with base packages for repositories and models. The setup is the same for the secondary data source, but with different prefixes. @Component @EnableJpaRepositories( basePackages = {"com.example.demo.models.primary", "com.example.demo.repositories.primary"}, entityManagerFactoryRef = "primaryEntityManagerFactory", transactionManagerRef = "primaryTransactionManager" ) public class DataSourcePrimaryManagers { @Primary @Bean("primaryEntityManagerFactory") public LocalContainerEntityManagerFactoryBean entityManagerFactory( @Qualifier(PRIMARY_DATASOURCE) DataSource dataSource) { LocalContainerEntityManagerFactoryBean em = new LocalContainerEntityManagerFactoryBean(); em.setDataSource(dataSource); em.setPackagesToScan("com.example.demo.models.primary", "com.example.demo.repositories.primary"); em.setJpaVendorAdapter(new HibernateJpaVendorAdapter()); return em; } @Primary @Bean(name = "primaryTransactionManager") public PlatformTransactionManager primaryTransactionManager(@Qualifier("primaryEntityManagerFactory") EntityManagerFactory entityManagerFactory) { return new JpaTransactionManager(entityManagerFactory); } } In the configuration above, we configured beans for JPA repositories and bound each to one of the data sources. The crucial step here is using JpaTransactionManager. This is important in terms of transaction awareness of any connection to a data source within an action annotated with Spring’s @Transactional.  Handling transactions for each data source will require explicitly defining transaction manager name @Transactional(transactionManager = PRIMARY_TRANSACTION_MANAGER) Read-only data source One of the useful multiple data sources use cases is introducing the read-only instance of the primary data source as the secondary data source.  The advantage of this reflects on performance - by redirecting the reading actions to the secondary data source and leaving the primary data source only in charge of other modifying actions will result in better throughput of the system and an overall performance boost compared to a system with the single data source.   If the secondary data source is unavailable for some reason, the system should be able to use only the primary data source. Therefore, here is the updated configuration with the support for this. @Configuration @ConfigurationProperties("spring.datasource.secondary") public class DataSourceSecondaryConfig extends HikariConfig { public static final String SECONDARY_DATASOURCE = "secondaryDataSource"; @Bean(SECONDARY_DATASOURCE) public HikariDataSource secondaryDataSource(@Qualifier(PRIMARY_DATASOURCE) HikariDataSource fallbackDataSource) { try { return new HikariDataSource(this); } catch (Exception e) { LOGGER.error("Exception happened while creating secondary data source: {}", e.getMessage(), e); return fallbackDataSource; } } } This is a simple yet incomplete usage of multiple data sources. For more complex scenarios, such as reading action failure due to overload or any technical issue, we need to consider transaction management across multiple data sources and handle switching back to the primary data source, for example.   Distributed transactions Handling distributed transactions is one of the most challenging cases with multiple data sources.  ChainedTransactionManager, a transaction manager implementation that was deprecated a few years ago, was trying to solve this use case where multiple transaction managers are involved. This implementation wasn’t robust enough, but fortunately, a lightweight library, Atomikos, relying on Java Transaction API (JTA) is a solution to this problem.  JTA is part of Java EE, and it is widely used for distributed transactions, especially when it comes to systems working with messaging and other sources. So far, in this article, we worked with local transactions bound to a single resource.In this context, a resource represents a data source, messaging service, or other data provider that engages with transactions. Further on, we will work with global transactions covering multiple resources participating in the transactional context. Atomikos incorporates JTA capabilities with Spring’s transaction management, allowing us to use it easily - with auto-configuration! For versions before Spring Boot 3, there is starter dependency by Spring spring-boot-starter-jta-atomikos, but for Spring Boot 3 is used Atomicos’ transactions-spring-boot3-starter:6.0.0. JTA context will be detected by using JTA-compliant data sources, and Spring will auto-configure transactionManager as JtaTransactionManager. Therefore, all resources taking place in a transactional context should be adapted for JTA. For example, the DataSource class we used for defining resources for local transactions should be replaced with the XADataSource class that is used for defining resources included in distributed transactions. In general, JTA relies on XA (Extended Architecture) standard, designed for distributed transactions, coordinating transactional actions from participating resources by a global transaction manager and ensuring atomicity of the main transaction. JTA transaction handling consists of two phase commit (2PC): - First of all, checking the status of commit action for both data sources - The second phase is determination based on those statuses. If one data source responds negatively, everything will be rolled back. Now, a few adaptations of the existing example are needed, starting from the configuration for the entity managers and repository references.  Adapting the previous data source configuration would look like this: @Configuration @ConfigurationProperties("spring.datasource.primary") public class DataSourcePrimaryConfig extends HikariConfig { public static final String PRIMARY_DATASOURCE = "primaryDataSource"; @Primary @Bean(PRIMARY_DATASOURCE) public DataSource primaryDataSource() { AtomikosDataSourceBean atomikosDataSource = new AtomikosDataSourceBean(); PGXADataSource pgxaDataSource = new PGXADataSource(); pgxaDataSource.setUrl(this.getJdbcUrl()); pgxaDataSource.setUser(this.getUsername()); pgxaDataSource.setPassword(this.getPassword()); atomikosDataSource.setUniqueResourceName(PRIMARY_DATASOURCE); atomikosDataSource.setXaDataSource(pgxaDataSource); atomikosDataSource.setMaxPoolSize(this.getMaximumPoolSize()); // Set the rest of the needed connection pool properties return atomikosDataSource; } } @Component @EnableJpaRepositories( basePackages = {"com.example.demo.models.primary", "com.example.demo.repositories.primary"}, entityManagerFactoryRef = "primaryEntityManagerFactory" ) public class DataSourcePrimaryManagers { @Primary @Bean("primaryEntityManagerFactory") public LocalContainerEntityManagerFactoryBean entityManagerFactory( @Qualifier(PRIMARY_DATASOURCE) DataSource dataSource) { LocalContainerEntityManagerFactoryBean em = new LocalContainerEntityManagerFactoryBean(); em.setDataSource(dataSource); em.setPackagesToScan("com.example.demo.models.primary", "com.example.demo.repositories.primary"); em.setJpaVendorAdapter(new HibernateJpaVendorAdapter()); Properties jpaProperties = new Properties(); jpaProperties.put("hibernate.current_session_context_class", "jta"); jpaProperties.put("javax.persistence.transactionType", "jta"); jpaProperties.put("hibernate.transaction.factory_class", "org.hibernate.transaction.JTATransactionFactory"); jpaProperties.put("hibernate.transaction.manager_lookup_class", "com.atomikos.icatch.jta.hibernate3.TransactionManagerLookup"); em.setJpaProperties(jpaProperties); return em; } } The changes to the secondary data source are made in the same manner. For data source bean is now used Atomiko’s class that implements XADataSource interface for distributed transaction, which interacts again with JDBC on the lower level. The organization of the properties is separated - PGXADataSource covers Postgres-related ones, and the rest is related to connection pool setup. For the entity manager, the only change is JPA and Hibernate setup, where the context is switched to JTA (setup provided by Atomiko’s docs - here). All transaction manager references should be omitted, since there will be only one transaction manager used, auto-configured by Spring.  Example Our data sources will both store user notes - the first will store personal notes and the other one work notes. We are keeping it simple, so the structure of both is the same, having id, note as textual field, created and updated date. The only difference between them is the UNIQUE constriction for the note field applied to work notes. Separate models named PersonalNote and WorkNote are defined in the separate packages primary and secondary within models, as mentioned above in the configuration for JPA repositories and the basePackages parameter. Here is an example of WorkNote class. @Entity @Table( name = "worknotes", uniqueConstraints = { @UniqueConstraint(columnNames = {"note"}) } ) public class WorkNote { @Id @GeneratedValue(strategy = GenerationType.IDENTITY) @Column(name = "id") private int id; @Column(name = "note") private String note; @Temporal(TemporalType.TIMESTAMP) @Column(name = "created") private Date created; @Temporal(TemporalType.TIMESTAMP) @Column(name = "updated") private Date updated; // Getters and setters } Now, the repository beans should be created. @Repository public interface PersonalNotesRepository extends JpaRepository<PersonalNote, Integer> { } @Repository public interface WorkNotesRepository extends JpaRepository<WorkNote, Integer> { } Moving forward to the service layer. Note that the transaction manager isn’t defined explicitly according to the used data source - it is referring to the global auto-configured transaction manager. @Service public class PersonalNotesService { private static final Logger LOGGER = LoggerFactory.getLogger(PersonalNotesService.class); private final PersonalNotesRepository personalNotesRepository; @Autowired public PersonalNotesService(PersonalNotesRepository personalNotesRepository) { this.personalNotesRepository = personalNotesRepository; } public List<PersonalNote> getAllPersonalNotes() { List<PersonalNote> notes = personalNotesRepository.findAll(); LOGGER.info("Returned {} personal note(s)", notes.size()); return notes; } @Transactional public List<PersonalNote> createNewPersonalNotes(final List<PersonalNote> newNotes) { List<PersonalNote> createdNotes = personalNotesRepository.saveAllAndFlush(newNotes); LOGGER.info("Created new {} personal note(s)", createdNotes.size()); return createdNotes; } @Transactional public List<PersonalNote> updatePersonalNotes(final List<PersonalNote> personalNotes) throws InterruptedException { List<PersonalNote> updatedNotes = personalNotesRepository.saveAllAndFlush(personalNotes); LOGGER.info("Updated {} personal note(s)", updatedNotes.size()); return updatedNotes; } } NotesService interacts with both data sources in transactional actions createNewNotes and updateNotes. @Service public class NotesService { private static final Logger LOGGER = LoggerFactory.getLogger(NotesService.class); private final PersonalNotesService personalNotesService; private final WorkNotesService workNotesService; @Autowired public NotesService(PersonalNotesService personalNotesService, WorkNotesService workNotesService) { this.personalNotesService = personalNotesService; this.workNotesService = workNotesService; } public Notes getAllNotes() { List<PersonalNote> personalNotes = personalNotesService.getAllPersonalNotes(); List<WorkNote> workNotes = workNotesService.getAllWorkNotes(); LOGGER.info("Returned {} personal note(s) and {} work note(s)", personalNotes.size(), workNotes.size()); return new Notes(personalNotes, workNotes); } @Transactional public Notes createNewNotes(final Notes notes) { List<PersonalNote> personalNotes = personalNotesService.createNewPersonalNotes(notes.getPersonalNotes()); LOGGER.info("Created {} personal note(s)", personalNotes.size()); List<WorkNote> workNotes = workNotesService.createNewWorkNotes(notes.getWorkNotes()); LOGGER.info("Created {} work note(s)", workNotes.size()); return new Notes(personalNotes, workNotes); } @Transactional public Notes updateNotes(final Notes notes) throws InterruptedException { List<PersonalNote> updatedPersonalNotes = personalNotesService.updatePersonalNotes(notes.getPersonalNotes()); LOGGER.info("Updated personal notes done at {}", new Date()); List<WorkNote> updatedWorkNotes = workNotesService.updateWorkNote(notes.getWorkNotes()); LOGGER.info("Updated work notes done at {}", new Date()); return new Notes(updatedPersonalNotes, updatedWorkNotes); } } When the createNewNotes action is executed, a new transaction will be created.  It will become a new participant in this transaction after establishing a connection to the first resource, primaryDataSource. Then, after the successful execution of the insert statement, a connection to the second resource, secondaryDataSource, will be established and added as a new participant to the existing transactional context.  Since the execution of the insert statement for the secondary resource wasn’t successful, the transaction will be completed by executing rollback for all participating resources. The important logs with bolded actions and information are shown below:  c.e.demo.controllers.NotesController : Requested creating notes o.s.t.jta.JtaTransactionManager : Creating new transaction with name [com.example.demo.services.NotesService.createNewNotes]: PROPAGATION_REQUIRED,ISOLATION_DEFAULT c.a.i.i.CompositeTransactionManagerImp : createCompositeTransaction ( 10000 ): created new ROOT transaction with id 127.0.0.1.tm172494476792000001 o.s.t.i.TransactionInterceptor : Getting transaction for [com.example.demo.services.NotesService.createNewNotes] o.s.t.jta.JtaTransactionManager : Participating in existing transaction o.s.t.i.TransactionInterceptor : Getting transaction for [com.example.demo.services.PersonalNotesService.createNewPersonalNotes] . . . connection logs for primary data source . . . c.a.icatch.imp.TransactionStateHandler : addParticipant ( XAResourceTransaction: XID: 3132372E302E302E312E7...) for transaction 127.0.0.1.tm172494476792000001 . . . execute insert statement logs . . . o.s.t.i.TransactionInterceptor : Completing transaction for [com.example.demo.services.PersonalNotesService.createNewPersonalNotes] com.example.demo.services.NotesService : Created 1 personal note(s) o.s.t.jta.JtaTransactionManager : Participating in existing transaction o.s.t.i.TransactionInterceptor : Getting transaction for [com.example.demo.services.WorkNotesService.createNewWorkNotes] . . . connection logs for secondary data source . . . c.a.icatch.imp.TransactionStateHandler : addParticipant ( XAResourceTransaction: XID: 3132372E302E302E312...) for transaction 127.0.0.1.tm172494476792000001 . . . execute insert statement logs . . . o.h.engine.jdbc.spi.SqlExceptionHelper : ERROR: duplicate key value violates unique constraint "unique_note" Detail: Key (note)=(new work note) already exists. c.a.icatch.imp.CompositeTransactionImp : setRollbackOnly() called for transaction 127.0.0.1.tm172494476792000001 o.s.t.jta.JtaTransactionManager : Participating transaction failed - marking existing transaction as rollback-only o.s.t.jta.JtaTransactionManager : Setting JTA transaction rollback-only o.s.t.i.TransactionInterceptor : Completing transaction for [com.example.demo.services.NotesService.createNewNotes] after exception: org.springframework.dao.DataIntegrityViolationException: could not execute statement [ERROR: duplicate key value violates unique constraint "unique_note" Detail: Key (note)=(new work note) already exists.] o.s.t.jta.JtaTransactionManager : Initiating transaction rollback c.a.datasource.xa.XAResourceTransaction : XAResource.rollback ( XID: 3132372E302E302E312E7...) on resource primaryDataSource represented by XAResource instance org.postgresql.xa.PGXAConnection@5f932d8e c.a.datasource.xa.XAResourceTransaction : XAResource.rollback ( XID: 3132372E302E302E312...) on resource secondaryDataSource represented by XAResource instance org.postgresql.xa.PGXAConnection@76571e69 c.a.icatch.imp.CompositeTransactionImp : rollback() done of transaction 127.0.0.1.tm172494476792000001 . . . closing connections after rollback logs . . . Conclusion As usual, Spring offers a few flexible options for transaction handling for multiple data sources, even when those transactions are dependent. A more robust and well-supported option would be Atomikos, which has auto-configuration and is easily integrated. Still, there is also the possibility to implement a custom transaction manager and take the matter in your own hands. Either way, this is not a simple task and should be approached with caution and testing that will require various cases.  *The DataSourceTransactionManager does not support true nested transactions by default. However, it supports savepoints, which allow partial rollbacks within a transaction if necessary, giving limited nested transaction-like behavior. https://docs.spring.io/spring-framework/reference/data-access/transaction/declarative/tx-propagation.html#tx-propagation-nested

Bucket Selector Aggregation Elasticsearch has many aggregations to offer that allow us to group and breakdown our data in almost any way we want. Often, the results of an aggregation have to be filtered based on different criteria that are calculated using that very same aggregation and then processed further. Sounds difficult? No worries, Elasticsearch has got that covered too. A pipeline aggregation called Bucket Selector comes to the rescue when filtering based on count, sum, or any other numeric value.  First, Bucket Aggregation has to be set-up for grouping documents into buckets, where each bucket is created for documents with a unique value of the given field. Then pipelining gathered buckets to Bucket Selector is the next step, where retrieving a certain subset of buckets is the main goal.  Example - Orders Index For this example, a simple orders index with mappings shown below will be used.  Mappings for this index produces the following response to the request GET orders/_mappings. { "orders" : { "mappings" : { "properties" : { "created_on" : {"type" : "date"}, "order_id" : {"type" : "long"}, "order_date" : {"type" : "date"}, "products" : { "properties" : { "product_id" : {"type" : "long"}, "product_name" : { "type" : "text", "fields" : { "keyword" : { "type" : "keyword", "ignore_above" : 256 } } }, "price" : {"type" : "float"} } }, "customer_id" : {"type" : "long"}, "customer_full_name" : { "type" : "text", "fields" : { "keyword" : { "type" : "keyword", "ignore_above" : 256 } } }, "shipping_country_code" : { "type" : "text", "fields" : { "keyword" : { "type" : "keyword", "ignore_above" : 256 } } }, "shipping_street" : {"type" : "text"}, "taxful_total_price" : {"type" : "float"}, "taxless_total_price" : {"type" : "float"}, ... } } } Let’s group some monthly orders into shipments - For each month of 2019, group orders by shipping country and select shipments (2500+ orders) with a total shipment price of over 300K.  Request GET orders/_search { "size": 0, "query": { // Filter all orders from 1st Jan 2019 - 1st Jan 2020 "range": { "order_date": { "gte": "2019-01-01T00:00:00.000Z", "lte": "2020-01-01T00:00:00.000Z" } } }, "aggs": { "orders_per_month": { // Group all orders per month "date_histogram": { "field": "order_date", "calendar_interval": "month" }, "aggs": { "orders": { // Group all orders in month per shipping country "terms": { "field": "shipping_country_code.keyword" }, "aggs": { "shipment_profit": { // Sum the prices as shipment_profit (this is per country, in a month) "sum": { "field": "taxful_total_price" } }, "country_orders_filter": { // Filter current buckets (current buckets are orders per country, per month) "bucket_selector": { "buckets_path": { "totalOrders": "_count", "totalProfit": "shipment_profit" }, "script": "params.totalOrders > 2500 && params.totalProfit > 300000" } } } } } } } } Note: Comments in the request are not part of the request, they just annotate sections of the query. "Elasticsearch - Bucket Selector Aggregation" Tech Bite was brought to you by Kristina Kraljević, Junior Software Engineer at Atlantbh. (more…)