You are viewing a plain text version of this content. The canonical link for it is here.
Posted to dev@kafka.apache.org by "Guozhang Wang (JIRA)" <ji...@apache.org> on 2016/02/04 17:29:39 UTC
[jira] [Assigned] (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:all-tabpanel ]
Guozhang Wang reassigned KAFKA-3207:
------------------------------------
Assignee: Guozhang Wang
> 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)