title: Google Pub/Sub to Dataflow, avoid duplicates with Record ID
body: <p>I'm trying to build a Streaming Dataflow Job which read events from Pub/Sub and write them into BigQuery.</p>
  
  <p>According to the documentation, Dataflow can detect duplicate messages delivery if a Record ID is used (see: <a href="https://cloud.google.com/dataflow/model/pubsub-io#using-record-ids" rel="nofollow noreferrer">https://cloud.google.com/dataflow/model/pubsub-io#using-record-ids</a>)</p>
  
  <p>But even using this Record ID, I still have some duplicates 
  (around 0.0002%).</p>
  
  <p>Did I miss something ?</p>
  
  <p>EDIT: </p>
  
  <p>I use <a href="https://github.com/spotify/async-google-pubsub-client" rel="nofollow noreferrer">Spotify Async PubSub Client</a> to publish messages with the following snipplet:</p>
  
  <pre><code>Message
        .builder()
        .data(new String(Base64.encodeBase64(json.getBytes())))
        .attributes("myid", id, "mytimestamp", timestamp.toString)
        .build()
  </code></pre>
  
  <p>Then I use <a href="https://github.com/spotify/scio" rel="nofollow noreferrer">Spotify scio</a> to read the message from pub/sub and save it to DataFlow:</p>
  
  <pre><code>val input = sc.withName("ReadFromSubscription")
                .pubsubSubscription(subscriptionName, "myid", "mytimestamp")
  input
      .withName("FixedWindow")
      .withFixedWindows(windowSize)  // apply windowing logic
      .toWindowed  // convert to WindowedSCollection
      //
      .withName("ParseJson")
      .map { wv =&gt;
        wv.copy(value = TableRow(
          "message_id" -&gt; (Json.parse(wv.value) \ "id").as[String],
          "message" -&gt; wv.value)
        )
      }
      //
      .toSCollection  // convert back to normal SCollection
      //
      .withName("SaveToBigQuery")
      .saveAsBigQuery(bigQueryTable(opts), BQ_SCHEMA, WriteDisposition.WRITE_APPEND)
  </code></pre>
  
  <p>The Window size is 1 minute.</p>
  
  <p>After only few seconds injecting messages I already have duplicates in BigQuery.</p>
  
  <p>I use this query to count duplicates:</p>
  
  <pre><code>SELECT 
     COUNT(message_id) AS TOTAL, 
     COUNT(DISTINCT message_id) AS DISTINCT_TOTAL 
  FROM my_dataset.my_table
  
  //returning 273666  273564
  </code></pre>
  
  <p>And this one to look at them:</p>
  
  <pre><code>SELECT *
  FROM my_dataset.my_table
  WHERE message_id IN (
    SELECT message_id
    FROM my_dataset.my_table
    GROUP BY message_id
    HAVING COUNT(*) &gt; 1
  ) ORDER BY message_id
  
  //returning for instance:
  row|id                                    | processed_at           | processed_at_epoch    
  1   00166a5c-9143-3b9e-92c6-aab52601b0be    2017-02-02 14:06:50 UTC 1486044410367   { ...json1... }  
  2   00166a5c-9143-3b9e-92c6-aab52601b0be    2017-02-02 14:06:50 UTC 1486044410368   { ...json1... }  
  3   00354cc4-4794-3878-8762-f8784187c843    2017-02-02 13:59:33 UTC 1486043973907   { ...json2... }  
  4   00354cc4-4794-3878-8762-f8784187c843    2017-02-02 13:59:33 UTC 1486043973741   { ...json2... } 
  5   0047284e-0e89-3d57-b04d-ebe4c673cc1a    2017-02-02 14:09:10 UTC 1486044550489   { ...json3... } 
  6   0047284e-0e89-3d57-b04d-ebe4c673cc1a    2017-02-02 14:08:52 UTC 1486044532680   { ...json3... }
  </code></pre>