|
| 1 | += CDC Synchronization of PostgreSQL and Elasticsearch with Spring Boot |
| 2 | + |
| 3 | +*Motivation* |
| 4 | + |
| 5 | +Some colleagues of mine use database and _Elasticsearch_(_ELK_) together where the database is the main data source, and _ELK_ is the search facade: the data is first stored in the database, and then the changes are written to the index. |
| 6 | + |
| 7 | +Here is a typical example of such code that uses _Spring Data_: |
| 8 | + |
| 9 | + @Override |
| 10 | + @Transactional |
| 11 | + public ProductDto update(ProductDto dto) { |
| 12 | + var jpaEntity = jpaRepository.save(dto2jpa(Objects.requireNonNull(dto))); |
| 13 | + var elkEntity = elkRepository.save(jpa2elk(jpaEntity)); |
| 14 | + return elk2dto(elkEntity); |
| 15 | + } |
| 16 | + |
| 17 | +Unfortunately, this approach does not guarantee the consistency of the database and the index, not only permanently, but also with a delay. Namely, there may be such a moment that the state of the database at that moment will not correspond to the state of the index at any moment. And vice versa. In particular, some changes in the database may never reach the index. |
| 18 | + |
| 19 | +Since I've seen code similar to the above in several projects, I thought a little project showing how sync could be done would be helpful. |
| 20 | + |
| 21 | +Thid is an example of such a _Java_ project using _Postgresql_ _CDC_(Change Data Capture), _Elasticsearch_, _String Boot_, _String Data_(_JPA_, _Elasticsearch_). The project does not use any external components such as _Kafka_, etc. Only the ones listed. There are ten classes in total. |
| 22 | + |
| 23 | +Below are brief comments on the project that do not fit into the self-describing code paradigm. |
| 24 | + |
| 25 | +*Brief digression* |
| 26 | + |
| 27 | +We will not touch on what _CDC_(Change Data Capture) is in general and its implementation in PostreSQL in particular. Details can be found at the link |
| 28 | + |
| 29 | +https://www.postgresql.org/docs/current/wal-intro.html[] |
| 30 | + |
| 31 | +Let's just say that this technology allows you to work with _WAL_ (Write-Ahead Logging) , which the database maintains to maintain data integrity and replication. This log records data about changes in the database, grouped by transactions. |
| 32 | + |
| 33 | +The main property of this log that can be used to synchronize the database and index is the following. |
| 34 | +Data in _WAL_ is stored exactly in the order in which it is written to the database. This allows you to write changes to the index exactly in the order in which they were written to the database, and not in the order in which control is transferred from _JPA_ component to _ELK_ one. |
| 35 | + |
| 36 | +Second, _CDC_ implementation differs between database vendors. The project shown uses PostgreSQL. |
| 37 | + |
| 38 | + |
| 39 | +*Configuration* |
| 40 | + |
| 41 | + |
| 42 | +In order to see how the project works, you must first set up the database. |
| 43 | + |
| 44 | +Open some client (_psql_, _DBeaver_, etc) and run the command |
| 45 | + |
| 46 | +`SHOW wal_level;` |
| 47 | + |
| 48 | +If the value is not "_logical_", then you need to execute the command |
| 49 | + |
| 50 | +`ALTER SYSTEM SET wal_level = logical;` |
| 51 | + |
| 52 | +Then you need to restart the db server or container with it ( at least for _PostgreSQL 14_). |
| 53 | + |
| 54 | +After that, open the client, look at the available slots |
| 55 | + |
| 56 | +`SELECT * FROM pg_replication_slots;` |
| 57 | + |
| 58 | +and create a slot with a unique name |
| 59 | + |
| 60 | +``SELECT * FROM pg_create_logical_replication_slot('elk_slot', 'test_decoding', false, true); |
| 61 | +`` |
| 62 | + |
| 63 | +The unique name of the slot (in the example '_elk_slot_') can be chosen by yourself. |
| 64 | + |
| 65 | +Let's leave the plugin name ('_test_decoding_') as it is for now. |
| 66 | + |
| 67 | +The rest of the settings can be left unchanged. |
| 68 | + |
| 69 | +In order to return to the previous configuration, you must delete the created slot |
| 70 | + |
| 71 | +`SELECT * FROM pg_drop_replication_slot ( 'elk_slot' );` |
| 72 | + |
| 73 | +restore the previous value of _wal_level_ |
| 74 | + |
| 75 | +`ALTER SYSTEM RESET wal_level;` |
| 76 | + |
| 77 | +and restart the server or the container again. |
| 78 | + |
| 79 | + |
| 80 | +*Project description* |
| 81 | + |
| 82 | + |
| 83 | +To make the project compact, but still fully functional, I limited myself to working with one entity - _Product_, which corresponds to the table of the same name in the database, the index of the same name and three Java classes: _ProductDto_, _JPA_ entity _ProductDB_ and _ELK_ document _ProductELK_. |
| 84 | + |
| 85 | +Two repositories are used to work with the entity |
| 86 | + |
| 87 | +`ProductJPARepository extends JpaRepository<ProductDB, Long>` |
| 88 | + |
| 89 | +``ProductELKRepository extends ElasticsearchRepository<ProductELK,Long> |
| 90 | +`` |
| 91 | + |
| 92 | +The _ProductVanillaService_ service works with these repositories. This service does not care about database and index synchronization. The above code is taken from this service. |
| 93 | + |
| 94 | +The controller that works with the service is _ProductController_, which has only _CRUD_ endpoints for brevity. |
| 95 | + |
| 96 | +The methods of the _ProductVanillaService_ service, as you can see from the example, are made transactional so that you can roll back changes in the database if there are problems with saving changes to the index. |
| 97 | + |
| 98 | +If the application is launched with the "_vanilla_" profile, then it will work without synchronization. |
| 99 | + |
| 100 | +In order to enable synchronization, you need to run the application with the “_sync_” profile. |
| 101 | + |
| 102 | +To keep the database and the index in sync _ProductCDCService_ service is used instead of _ProductVanillaService_. The sych version uses the same repositories as the vanilla one. |
| 103 | + |
| 104 | +Here is an example of processing data changes from ProductCDCService, similar to the example given at the beginning of the article. |
| 105 | + |
| 106 | + @Override |
| 107 | + @Transactional(propagation= Propagation.NEVER) |
| 108 | + public ProductDto update(ProductDto dto) { |
| 109 | + var id = Objects.requireNonNull(Objects.requireNonNull(dto).getId()); |
| 110 | + jpaRepository.save(dto2jpa(dto)); |
| 111 | + processNextCDCChunk(); |
| 112 | + return elkRepository.findById(id).map(ProductMapper::elk2dto).orElse(null); |
| 113 | + } |
| 114 | + |
| 115 | +Working with the database for the two services is identical. Except that the methods of the first service are transactional and therefore saving data to the index occurs in the same transaction as saving data to the database. |
| 116 | + |
| 117 | +The methods of the _ProductCDCService_ service are not transactional. |
| 118 | + |
| 119 | +Also, the _ProductCDCService_ methods, unlike _ProductVanillaService_, do not store data in the index. |
| 120 | +Instead, it calls _processNextCDCChunk()_ method, which starts scanning completed transactions from _WAL_. |
| 121 | + |
| 122 | +This method starts its work outside a transaction. The transaction is opened in |
| 123 | + |
| 124 | +`dbRepository.save(dto2db(dto))` |
| 125 | + |
| 126 | +and closed on exit from it. |
| 127 | + |
| 128 | +Therefore, by the time _processNextCDCChunk()_ is called, the data saved in the previous step is already in _WAL_ and available for scanning by this method. And, as a result, by the time _processNextCDCChunk()_ ends, this data will already be in the index. |
| 129 | + |
| 130 | +With the next call to elkRepository.findById(id) we get this data from the index and return it to the controller. |
| 131 | + |
| 132 | +Here the next question arises. Does the state of the returned object match the state the object was at call of _update(dto)_? For example, if user A changed the product name to "_prodA_", will the name stay that way in the output? |
| 133 | +The answer to this question is negative. |
| 134 | +The correct answer is the following. |
| 135 | +We return the state of the product at SOME point in time after the database transaction was committed. |
| 136 | + |
| 137 | +Let's take an example. |
| 138 | +Suppose user A renamed the product to "prodA" and user B followed him to "_prodB_". |
| 139 | +If by the time _processNextCDCChunk()_ method of user A completed WAL scan, the transaction with user B's change has been committed, then the result of this commit will be included in the scan of user A, and user A's _elkRepository.findById(id)_ will return the product with the value of the field name "_prodB_". |
| 140 | + |
| 141 | +There is nothing strange in this, as it is clear that the state of the database is saved in the index with some delay. The main thing is that all database changes reach the index. |
| 142 | + |
| 143 | +In particular, after successfully saving changes to the database and to the index, we may not return anything to the user. This can happen if, following the changes that we process in client A's transaction, this object is deleted in client B's transaction. And if by the time user A scans _WAL_, the deletion is already committed, then user A's _processNextCDCChunk()_ will remove the object from the index. In this case, upon return from _processNextCDCChunk()_, this object will no longer be in the index. Unless, of course, some third user C re-createed the deleted product and this productgets into the index by the time _elkRepository.findById(id)_ of user A is called. In this case, we will return the product of user C to the user A, not the product of user A with with the name "_prodA_". |
| 144 | + |
| 145 | + |
| 146 | + |
| 147 | +The _processNextCDCChunk()_ method is simple. It calls the _TestDecodingCDCService.processNextCDCChunk()_ method and handles the exception. It is important to note two points here. |
| 148 | + |
| 149 | +The first is that the method is called asynchronously. |
| 150 | + |
| 151 | +And the second is that the executor that executes the method uses one thread. This ensures that _WAL_ is processed sequentially, and therefore data changes are loaded into the index in exactly the same order as those changes were loaded into the database. Therefore, the state of the index at each moment will correspond to the state of the database at some moment in the past. |
| 152 | + |
| 153 | + |
| 154 | +_TestDecodingCDCService.processNextCDCChunk()_ is the central method. It takes all committed unprocessed transactions, parses and uploads changes from scanned transactions into the index. |
| 155 | + |
| 156 | +Uploading changes to the index (_TransactionOperationProcessor.processOp()_) consists of the following steps: |
| 157 | + |
| 158 | +1. Based on the table name (_TransactionOperation.tableName_) from the transaction operation, determine the _JPA_ entity class that is persisted in this table. If the class is not defined, then the operation is skipped. |
| 159 | + |
| 160 | +2. Based on the _JPA_ class, find the _ELK_ service that processes the entities of this class. This service wraps _ProductELKRepository_. _ProductVanillaService_ uses this particular repository to upload _JPA_ entity into the index. If no such service is found, then the operation is skipped. |
| 161 | + |
| 162 | +3. Using _JPA_, restore the _JPA_ entity by _TransactionOperation.restoreSQLStatement_. Since the operation is associated with a specific table (_TransactionOperation.tableName_), only that part of the entity that is persisted in this table is restored. _JPA_ properties annotated with _@OneToOne_, _@OneToMany_, etc. are not initialized. |
| 163 | + |
| 164 | +4. Upload the restored _JPA_ entity into the index with the service found in step 2. |
| 165 | + |
| 166 | + |
| 167 | +If there are any problems during the processing of operations, you need to fix the problems and re-run _TestDecodingCDCService.processNextCDCChunk()_. |
| 168 | +This can be done because the _ProductELKRepository.save_ and _ProductELKRepository.delete_ are idempotent. |
| 169 | + |
| 170 | +If all scanned transactions are successfully processed, then the corresponding records are removed from _WAL_. |
| 171 | + |
| 172 | +I hope provided information is helpful. |
| 173 | + |
| 174 | +Please contact us if you have any comments, suggestions or questions. |
| 175 | + |
| 176 | +Stay in sync, |
| 177 | + |
| 178 | +Sergey |
| 179 | + |
| 180 | + |
| 181 | + |
| 182 | + |
| 183 | + |
| 184 | + |
0 commit comments