You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "ASF GitHub Bot (JIRA)" <ji...@apache.org> on 2016/02/04 19:14:39 UTC
[jira] [Commented] (KAFKA-3207) StateStore seems to be writing
state to one topic but restoring from another
[ 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 <wa...@gmail.com>
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)