title: How to configure a Kafka topic for use as snapshot store
body: <p>This topic should hold only the latest 'document X updated' event for each X. But I cannot get the topic configured correctly and it retains multiple copies. </p>
  
  <p>My idea is to keep segments small, as well as all related timeouts, flush and retention times. </p>
  
  <p>The topic settings (I do not have a clear enough understanding of where and with what prefix each of the options is applied, so there are probably several unused and unrelated ones as well as exaggerated numbers - corrections welcome):</p>
  
  <pre><code>"cleanup.policy"                    -&gt; "compact",
  "file.delete.delay.ms"              -&gt; "10",
  "segment.bytes"                     -&gt; "10000",
  "delete.retention.ms"               -&gt; "10",
  "retention.bytes"                   -&gt; "10000",
  "segment.ms"                        -&gt; "10",
  "retention.ms"                      -&gt; "10",
  "min.cleanable.dirty.ratio"         -&gt; "0.001",
  "flush.messages"                    -&gt; "1",
  "flush.ms"                          -&gt; "10",
  "min.compaction.lag.ms"             -&gt; "1",
  "log.cleaner.min.compaction.lag.ms" -&gt; "1" 
  </code></pre>
  
  <p>I feed the topic with <a href="http://doc.akka.io/docs/akka-stream-kafka/current/home.html" rel="nofollow noreferrer">akka-streams-kafka</a>: </p>
  
  <pre><code>val ids = List("12345", ...)
  
  val publish: Future[Done] = Source(ids ++ ids ++ ids ++ ids ++ ids)
    .map { id =&gt;
      ProducerMessage.Message(new ProducerRecord[String, String](topic, id, id), id)
    }
    .via(producerFlow)
    .map(logResult)
    .runWith(Sink.ignore)
  Await.result(publish, 3.seconds)
  </code></pre>
  
  <p>after waiting for several seconds, I count the messages: </p>
  
  <pre><code>var count = 0
  val runCount = Consumer
    .plainSource(consumerSettings, Subscriptions.topics(topic))
    .map { t =&gt;
      count += 1
      t
    }
    .runWith(Sink.ignore)
  Try { Await.result(runCount, timeout) }
  </code></pre>
  
  <p>I expect the consumer to receive <code>ids.length</code> messages, but it always receives all the produced messages on the first run and even more on subsequent runs.</p>
  
  <p>Some compaction does take place - if I run the test several times, the count of consumed messages stops growing and I see the segment deletions in the kafka log - but there are still multiple messages for each key. </p>
  
  <p>How do I use a kafka topic as a snapshot store?</p>
  
  <p>Using kafka 0.10.2.1</p>
  
  <p>Thank you.</p>