title: Getting 409 from the schema registry while saving records to a local state store where multiple state stores are associated with a single processor
body: <p>Long story short: I am in the middle of implementing a processor topology: the processor is to store the received records into corresponding local state stores and do event-based processing as a record arrives. And the related code looks like this:</p>
  <pre><code>    @Override
    public void configureBuilder(StreamsBuilder builder) {
      final Map&lt;String, String&gt; serdeConfig =
          Collections.singletonMap(&quot;schema.registry.url&quot;, processorConfig.getSchemaRegistryUrl());
  
      final Serde&lt;GenericRecord&gt; valueSerde = new GenericAvroSerde();
      valueSerde.configure(serdeConfig, false); // `true` for record keys
      final Serde&lt;EventKey&gt; keySerde = new SpecificAvroSerde();
      keySerde.configure(serdeConfig, true); // `true` for record keys
  
      Map&lt;String, String&gt; stateStoreConfigMap = new HashMap&lt;&gt;();
      //stateStoreConfigMap.put(KafkaAvroSerializerConfig.VALUE_SUBJECT_NAME_STRATEGY, RecordNameStrategy.class.getName());
      StoreBuilder&lt;KeyValueStore&lt;EventKey, GenericRecord&gt;&gt; aggSequenceStateStoreBuilder =
          Stores.keyValueStoreBuilder(
                  Stores.persistentKeyValueStore(processStateStore), keySerde, valueSerde)
              .withLoggingEnabled(stateStoreConfigMap)
              .withCachingEnabled();
  
      final Serde&lt;EnrichedSmcHeatData&gt; enrichedSmcHeatDataSerde = new SpecificAvroSerde&lt;&gt;();
      enrichedSmcHeatDataSerde.configure(serdeConfig, false); // `true` for record keys
      StoreBuilder&lt;KeyValueStore&lt;EventKey, EnrichedSmcHeatData&gt;&gt; enrichedSmcHeatStateStoreBuilder =
          Stores.keyValueStoreBuilder(
              Stores.persistentKeyValueStore(&quot;enriched-smc-heat-state-store&quot;), keySerde, enrichedSmcHeatDataSerde)
              .withLoggingEnabled(stateStoreConfigMap)
              .withCachingEnabled();
  
      Topology topology = builder.build();
      topology
          .addSource(
              PROCESS_EVENTS_SOURCE,
              keySerde.deserializer(),
              valueSerde.deserializer(),
              processorConfig.getInputCcmProcessEvents())
          .addSource(
              SCHEDULED_SEQUENCES_SOURCE,
              keySerde.deserializer(),
              valueSerde.deserializer(),
              processorConfig.getScheduledCastSequences())
          .addSource(
              SMC_HEAT_EVENTS_SOURCE,
              keySerde.deserializer(),
              valueSerde.deserializer(),
              processorConfig.getInputSmcHeatEvents())
          .addProcessor(
              PROCESS_STATE_AGGREGATOR,
              () -&gt; new ProcessStateProcessor(processStateStore, processorConfig),
              PROCESS_EVENTS_SOURCE,
              SCHEDULED_SEQUENCES_SOURCE,
              SMC_HEAT_EVENTS_SOURCE)
          .addStateStore(aggSequenceStateStoreBuilder, PROCESS_STATE_AGGREGATOR)
          .addStateStore(enrichedSmcHeatStateStoreBuilder, PROCESS_STATE_AGGREGATOR);
  </code></pre>
  <p>If there are updates for the store created by aggSequenceStateStoreBuilder, the record values could be saved to the store without problems. However, if updates came for the second store, the following error was getting thrown:</p>
  <blockquote>
  <p>Caused by:
  io.confluent.kafka.schemaregistry.client.rest.exceptions.RestClientException:
  Schema being registered is incompatible with an earlier schema for
  subject
  &quot;ccm-process-events-processor-ccm-process-state-store-changelog-value&quot;;
  error code: 409</p>
  </blockquote>
  <p>My use case: the state processor accepts inbound records from multiple source topics and do event-handling (including storing the modified values to the corresponding stores) when a record arrives from any of the source topics.</p>
  <p>It appears that there can only be one schema registered with the schema registry for the same processor. Is that by design, or am I missing anything, or what alternative options do I have instead?</p>
  <p>Thanks in advance!</p>