[
https://issues.apache.org/jira/browse/KAFKA-3207?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15132692#comment-15132692
]
ASF GitHub Bot commented on KAFKA-3207:
---------------------------------------
GitHub user guozhangwang opened a pull request:
https://github.com/apache/kafka/pull/865
KAFKA-3207: Fix StateChangeLogger to use the right topic name
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/guozhangwang/kafka K3207
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/kafka/pull/865.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 #865
----
commit 0e6392cfa51d9ba0cc1abb9052abdff3316cec74
Author: Guozhang Wang <[email protected]>
Date: 2016-02-04T18:13:05Z
fis StateChangeLogger to use the right topic name
----
> StateStore seems to be writing state to one topic but restoring from another
> ----------------------------------------------------------------------------
>
> Key: KAFKA-3207
> URL: https://issues.apache.org/jira/browse/KAFKA-3207
> Project: Kafka
> Issue Type: Bug
> Components: kafka streams
> Affects Versions: 0.9.1.0
> Environment: MacOS El Capitan
> Reporter: Tom Dearman
> Assignee: Guozhang Wang
> Priority: Blocker
>
> The state store (I am using in-memory state store) writes to a topic call
> [store-name] but restores from [job-id]-[store-name]-changelog. You can see
> in StoreChangeLogger that it writes to a topic which is the [store-name]
> passed through from the store supplier factory, but restores from the above
> topic name. My topology is:
> TopologyBuilder builder = new TopologyBuilder();
> SerializerAdapter<CommonKey> commonKeyAdapter = new
> SerializerAdapter<>(JDKBinarySerializer.INSTANCE);
> SerializerAdapter<GamePlayValue> gamePlayAdapter = new
> SerializerAdapter<>(JDKBinarySerializer.INSTANCE);
> builder.addSource("SOURCE", commonKeyAdapter, gamePlayAdapter,
> kafkaStreamConfig.getGamePlayTopic());
> Duration activityInterval =
> kafkaStreamConfig.getActivityInterval();
> if (activityInterval.toMinutes() % 5 != 0 || 24 * 60 %
> activityInterval.toMinutes() != 0)
> {
> throw new SystemFaultException(
> "The game activity interval must be a multiple
> of 5 minutes and divide into 24 hours current value [" +
> activityInterval.toMinutes() + "]");
> }
> builder.addProcessor("PROCESS", new
> GameActivitySupplier(kafkaStreamConfig.getStoreName(),
>
> kafkaStreamConfig.getGameActivitySendPeriod(),
>
> activityInterval,
>
> kafkaStreamConfig.getRemoveOldestTime(),
>
> kafkaStreamConfig.getRemoveAbsoluteTime()), "SOURCE");
> SerializerAdapter<StoreValue> storeValueAdapter = new
> SerializerAdapter<>(JDKBinarySerializer.INSTANCE);
> builder.addStateStore(
>
> Stores.create(kafkaStreamConfig.getStoreName()).withKeys(commonKeyAdapter,
> commonKeyAdapter).withValues(
> storeValueAdapter,
> storeValueAdapter).inMemory().build(), "PROCESS");
> builder.addSink("SINK",
> kafkaStreamConfig.getGameActivityTopic(), commonKeyAdapter,
> new
> SerializerAdapter<GameActivityTotalMessage>(JDKBinarySerializer.INSTANCE),
> "PROCESS");
--
This message was sent by Atlassian JIRA
(v6.3.4#6332)
