valitydev/wb-list-manager
14
0
Fork 0
mirror of https://github.com/valitydev/wb-list-manager.git synced 2024-11-06 01:35:17 +00:00
Code Issues Actions Packages Projects Releases Wiki Activity

Fix stream (#26)

Browse Source
This commit is contained in:
struga 2023-01-26 18:01:23 +07:00 committed by GitHub
parent 9c8ac08b05
commit e9b6d5b7be
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 5 additions and 3 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

8
src/main/java/dev/vality/wb/list/manager/listener/StartupListener.java
View File

@ -35,10 +35,12 @@ public class StartupListener implements ApplicationListener<ContextRefreshedEven
log.info("StartupListener start stream kafkaStreams: {}", kafkaStreams.metadataForAllStreamsClients());
if (wbListCorrectionStreamProperties.getEnabled()) {
wbListStreamProperties.put("application.id", wbListCorrectionStreamProperties.getApplicationId());
wbListStreamProperties.put("client.id", wbListCorrectionStreamProperties.getClientId());
Properties properties = new Properties();
properties.putAll(wbListStreamProperties);
properties.put("application.id", wbListCorrectionStreamProperties.getApplicationId());
properties.put("client.id", wbListCorrectionStreamProperties.getClientId());
kafkaStreamsWbListErrorRowsCorrection =
wbListErrorRowsCorrectionStreamFactory.create(wbListStreamProperties);
wbListErrorRowsCorrectionStreamFactory.create(properties);
kafkaStreamsWbListErrorRowsCorrection.start();
log.info("StartupListener start stream kafkaStreamsWbListErrorRowsCorrection: {}",
kafkaStreamsWbListErrorRowsCorrection.metadataForAllStreamsClients());