GitHub user dguy opened a pull request:
https://github.com/apache/kafka/pull/2051
KAFKA-4311: Multi layer cache eviction causes forwarding to incorrect
ProcessorNode
Given a topology like the one below. If a record arriving in `tableOne`
causes a cache eviction, it will trigger the `leftJoin` that will do a `get`
from `reducer-store`. If the key is not currently cached in `reducer-store`,
but is in the backing store, it will be put into the cache, and it may also
trigger an eviction. If it does trigger an eviction and the eldest entry is
dirty it will flush the dirty keys. It is at this point that a
ClassCastException is thrown. This occurs because the ProcessorContext is still
set to the context of the `leftJoin` and the next child in the topology is
`mapValues`.
We need to set the correct `ProcessorNode`, on the context, in the
`ForwardingCacheFlushListener` prior to calling `context.forward`. We also need
to set remember to reset the `ProcessorNode` to the previous node once
`context.forward` has completed.
```
final KTable<String, String> one = builder.table(Serdes.String(),
Serdes.String(), tableOne, tableOne);
final KTable<Long, String> two = builder.table(Serdes.Long(),
Serdes.String(), tableTwo, tableTwo);
final KTable<String, Long> reduce = two.groupBy(new
KeyValueMapper<Long, String, KeyValue<String, Long>>() {
@Override
public KeyValue<String, Long> apply(final Long key, final
String value) {
return new KeyValue<>(value, key);
}
}, Serdes.String(), Serdes.Long())
.reduce(new Reducer<Long>() {..}, new Reducer<Long>() {..},
"reducer-store");
one.leftJoin(reduce, new ValueJoiner<String, Long, String>() {..})
.mapValues(new ValueMapper<String, String>() {..});
```
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/dguy/kafka kafka-4311
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/2051.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #2051
----
commit 5c8896a9d2aaca6bf5b9fb8de4de8140919fb280
Author: Damian Guy <damian....@gmail.com>
Date: 2016-10-21T11:27:17Z
Save and set the current processor node in FlushingCacheListener.
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastruct...@apache.org or file a JIRA ticket
with INFRA.
---
