How can i determine that a Keystore is updated in Spring Cloud Stream API
.everyoneloves__top-leaderboard:empty,.everyoneloves__mid-leaderboard:empty,.everyoneloves__bot-mid-leaderboard:empty{ height:90px;width:728px;box-sizing:border-box;
}
0
I create a materialized view from a topic with the aggregate function from the spring cloud stream api.
This looks like the following:
public void process(KStream<Object, Object> input){
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
Then i query my created Statestore with:
ReadOnlyWindowStore<Object, Object> windowStore =
queryService.getQueryableStoreType("test", QueryableStoreTypes.windowStore());
Now my question is how can i determine that this statestore has updated after a new event was handled by the process method? Is their somekind of event i can listen to or can i create one?
spring apache-kafka spring-cloud apache-kafka-streams spring-cloud-stream
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
add a comment |
0
I create a materialized view from a topic with the aggregate function from the spring cloud stream api.
This looks like the following:
public void process(KStream<Object, Object> input){
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
Then i query my created Statestore with:
ReadOnlyWindowStore<Object, Object> windowStore =
queryService.getQueryableStoreType("test", QueryableStoreTypes.windowStore());
Now my question is how can i determine that this statestore has updated after a new event was handled by the process method? Is their somekind of event i can listen to or can i create one?
spring apache-kafka spring-cloud apache-kafka-streams spring-cloud-stream
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
add a comment |
0
0
0
I create a materialized view from a topic with the aggregate function from the spring cloud stream api.
This looks like the following:
public void process(KStream<Object, Object> input){
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
Then i query my created Statestore with:
ReadOnlyWindowStore<Object, Object> windowStore =
queryService.getQueryableStoreType("test", QueryableStoreTypes.windowStore());
Now my question is how can i determine that this statestore has updated after a new event was handled by the process method? Is their somekind of event i can listen to or can i create one?
spring apache-kafka spring-cloud apache-kafka-streams spring-cloud-stream
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
I create a materialized view from a topic with the aggregate function from the spring cloud stream api.
This looks like the following:
public void process(KStream<Object, Object> input){
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
Then i query my created Statestore with:
ReadOnlyWindowStore<Object, Object> windowStore =
queryService.getQueryableStoreType("test", QueryableStoreTypes.windowStore());
Now my question is how can i determine that this statestore has updated after a new event was handled by the process method? Is their somekind of event i can listen to or can i create one?
spring apache-kafka spring-cloud apache-kafka-streams spring-cloud-stream
spring apache-kafka spring-cloud apache-kafka-streams spring-cloud-stream
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
asked Nov 23 '18 at 13:26
axelDitertaaxelDiterta
496
496
add a comment |
add a comment |
1 Answer
1
active
oldest
votes
0
Your program is:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
In fact, the last aggregate() returns a KTable object. If you disable caching via Materialized you can get informed about every single update to the KTable via:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized) // disable caching via Materialized
.toStream()
.foreach(...) // react to every update to the KTable
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates intoforeach()(cf. docs.confluent.io/current/streams/developer-guide/…)
– Matthias J. Sax
Nov 26 '18 at 16:53
add a comment |
Your Answer
StackExchange.ifUsing("editor", function () {
StackExchange.using("externalEditor", function () {
StackExchange.using("snippets", function () {
StackExchange.snippets.init();
});
});
}, "code-snippets");
StackExchange.ready(function() {
var channelOptions = {
tags: "".split(" "),
id: "1"
};
initTagRenderer("".split(" "), "".split(" "), channelOptions);
StackExchange.using("externalEditor", function() {
// Have to fire editor after snippets, if snippets enabled
if (StackExchange.settings.snippets.snippetsEnabled) {
StackExchange.using("snippets", function() {
createEditor();
});
}
else {
createEditor();
}
});
function createEditor() {
StackExchange.prepareEditor({
heartbeatType: 'answer',
autoActivateHeartbeat: false,
convertImagesToLinks: true,
noModals: true,
showLowRepImageUploadWarning: true,
reputationToPostImages: 10,
bindNavPrevention: true,
postfix: "",
imageUploader: {
brandingHtml: "Powered by u003ca class="icon-imgur-white" href="https://imgur.com/"u003eu003c/au003e",
contentPolicyHtml: "User contributions licensed under u003ca href="https://creativecommons.org/licenses/by-sa/3.0/"u003ecc by-sa 3.0 with attribution requiredu003c/au003e u003ca href="https://stackoverflow.com/legal/content-policy"u003e(content policy)u003c/au003e",
allowUrls: true
},
onDemand: true,
discardSelector: ".discard-answer"
,immediatelyShowMarkdownHelp:true
});
}
});
draft saved
draft discarded
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53447573%2fhow-can-i-determine-that-a-keystore-is-updated-in-spring-cloud-stream-api%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
1 Answer
1
active
oldest
votes
1 Answer
1
active
oldest
votes
active
oldest
votes
active
oldest
votes
0
Your program is:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
In fact, the last aggregate() returns a KTable object. If you disable caching via Materialized you can get informed about every single update to the KTable via:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized) // disable caching via Materialized
.toStream()
.foreach(...) // react to every update to the KTable
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates intoforeach()(cf. docs.confluent.io/current/streams/developer-guide/…)
– Matthias J. Sax
Nov 26 '18 at 16:53
add a comment |
0
Your program is:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
In fact, the last aggregate() returns a KTable object. If you disable caching via Materialized you can get informed about every single update to the KTable via:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized) // disable caching via Materialized
.toStream()
.foreach(...) // react to every update to the KTable
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates intoforeach()(cf. docs.confluent.io/current/streams/developer-guide/…)
– Matthias J. Sax
Nov 26 '18 at 16:53
add a comment |
0
0
0
Your program is:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
In fact, the last aggregate() returns a KTable object. If you disable caching via Materialized you can get informed about every single update to the KTable via:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized) // disable caching via Materialized
.toStream()
.foreach(...) // react to every update to the KTable
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
Your program is:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized)
In fact, the last aggregate() returns a KTable object. If you disable caching via Materialized you can get informed about every single update to the KTable via:
input
.peek((key, value) ->{...}
.map((key, value) -> {...}
.groupByKey()
.windowedBy(TimeWindows.of(5000))
.aggregate(Initializer, Aggregator, Materialized) // disable caching via Materialized
.toStream()
.foreach(...) // react to every update to the KTable
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
answered Nov 23 '18 at 19:07
Matthias J. SaxMatthias J. Sax
31.9k45583
31.9k45583
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates intoforeach()(cf. docs.confluent.io/current/streams/developer-guide/…)
– Matthias J. Sax
Nov 26 '18 at 16:53
add a comment |
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates intoforeach()(cf. docs.confluent.io/current/streams/developer-guide/…)
– Matthias J. Sax
Nov 26 '18 at 16:53
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
Is it possible to do both at once? Or should i create 2 different aggregates , one for caching and one for listening and then triggering something?
– axelDiterta
Nov 26 '18 at 10:54
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates into
foreach() (cf. docs.confluent.io/current/streams/developer-guide/…)– Matthias J. Sax
Nov 26 '18 at 16:53
"Caching" is an overloaded term here... If you update the code as suggested, you still get your materialized view and you can still query it. "Caching" means something different and is related to downstream push updates into
foreach() (cf. docs.confluent.io/current/streams/developer-guide/…)– Matthias J. Sax
Nov 26 '18 at 16:53
add a comment |
draft saved
draft discarded
Thanks for contributing an answer to Stack Overflow!
- Please be sure to answer the question. Provide details and share your research!
But avoid …
- Asking for help, clarification, or responding to other answers.
- Making statements based on opinion; back them up with references or personal experience.
To learn more, see our tips on writing great answers.
draft saved
draft discarded
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
StackExchange.ready(
function () {
StackExchange.openid.initPostLogin('.new-post-login', 'https%3a%2f%2fstackoverflow.com%2fquestions%2f53447573%2fhow-can-i-determine-that-a-keystore-is-updated-in-spring-cloud-stream-api%23new-answer', 'question_page');
}
);
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Sign up or log in
StackExchange.ready(function () {
StackExchange.helpers.onClickDraftSave('#login-link');
});
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Sign up using Google
Sign up using Facebook
Sign up using Email and Password
Post as a guest
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
Required, but never shown
