DB Oracle. Before multiple instances was only one instance. And problem with repeated read was solved with "select .. for update skip locked".
JdbcPollingChannelAdapter adapter = new JdbcPollingChannelAdapter(
dataSource, SELECT_QUERY);
adapter.setUpdateSql(UPDATE_QUERY);
adapter.setUpdateSqlParameterSourceFactory(input -> {
List<TableEntry> inputList = (List<TableEntry>) input;
List<String> ids = inputList.stream()
.map(TableEntry::getId)
.collect(ImmutableList.toImmutableList());
return new MapSqlParameterSource()
.addValue("ids", ids)
.addValue("status", TableStatus.IN_PROGRESS);
});
adapter.setRowMapper((rs, rowNum) ->
TableEntry.builder()
.id(rs.getString("id"))
...
...
.status(rs.getString("status"))
.build());
String UPDATE_QUERY = "UPDATE my_table "
+ "SET status = :status "
+ "WHERE id in (:ids)";
String SELECT_QUERY = String.format(""
+ "select * from my_table "
+ "where id in "
+ " (select id from "
+ " (select * from my_table "
+ " where status = '%s' "
+ " order by id asc)"
+ " where rowNum <= 10)"
+ "for update skip locked", TableStatus.NEW);
And using that poller in IntegrationFlow:
return IntegrationFlows
.from(pollingAdapter, c -> c.poller(Pollers
.fixedRate(updateRateTime, TimeUnit.MILLISECONDS)
.transactional()
.maxMessagesPerPoll(messagesPerPoll)))
.split()
...
...
But recently we have had the need to have two instances. And we have problems with repeated read. Both nodes read the same and we have duplicates in process. And it is necessary that the nodes read different records.
I thought about serializable isolation level in transaction but it's too expensive. Maybe there is a way to make it with optimistic lock. Or maybe thereis a way to make lock inside Oracle. Or something else) Thanks!