title: Google Cloud Pubsub Data lost
body: <p>I'm experiencing a problem with GCP pubsub where a small percentage of data was lost when publishing thousands of messages in couple seconds. </p>
  
  <p>I'm logging both <code>message_id</code> from pubsub and a <code>session_id</code> unique to each message on both the publishing end as well as the receiving end, and the result I'm seeing is that some message on the receiving end has same <code>session_id</code>, but different <code>message_id</code>. Also, some messages were missing.</p>
  
  <p>For example, in one test I send 5,000 messages to pubsub, and exactly 5,000 messages were received, with 8 messages lost. The log lost messages look like this:</p>
  
  <pre><code>MISSING sessionId:sessionId: 731 (missing in log from pull request, but present in log from Flask API)
  
  messageId FOUND: messageId:108562396466545
  
  API: 200 **** sessionId: 731, messageId:108562396466545 ******(Log from Flask API)
  
  Pubsub: sessionId: 730, messageId:108562396466545(Log from pull request)
  </code></pre>
  
  <p>And the duplicates looks like: </p>
  
  <pre><code>======= Duplicates FOUND on sessionId: 730=======
  
  sessionId: 730, messageId:108562396466545
  
  sessionId: 730, messageId:108561339282318
  
  (both are logs from pull request)
  </code></pre>
  
  <p>All missing data and duplicates look like this.</p>
  
  <p>From the above example, it is clear that some messages has taken the <code>message_id</code> of another message, and has been sent twice with two different <code>message_id</code>s.</p>
  
  <p>I wonder if anyone would help me figure out what is going on? Thanks in advance.</p>
  
  <p><strong>Code</strong></p>
  
  <p>I have an API sending message to pubsub, which looks like this:</p>
  
  <pre class="lang-python prettyprint-override"><code>from flask import Flask, request, jsonify, render_template
  from flask_cors import CORS, cross_origin
  import simplejson as json
  from google.cloud import pubsub
  from functools import wraps
  import re
  import json
  
  
  app = Flask(__name__)
  ps = pubsub.Client()
  
  ...
  
  @app.route('/publish', methods=['POST'])
  @cross_origin()
  @json_validator
  def publish_test_topic():
      pubsub_topic = 'test_topic'
      data = request.data
  
      topic = ps.topic(pubsub_topic)
  
      event = json.loads(data)
  
      messageId = topic.publish(data)
      return '200 **** sessionId: ' + str(event["sessionId"]) + ", messageId:" + messageId + " ******"
  </code></pre>
  
  <p>And this is the code I used to read from pubsub:</p>
  
  <p>from google.cloud import pubsub
      import re
      import json</p>
  
  <pre><code>ps = pubsub.Client()
  topic = ps.topic('test-xiu')
  sub = topic.subscription('TEST-xiu')
  
  max_messages = 1
  stop = False
  
  messages = []
  
  class Message(object):
      """docstring for Message."""
      def __init__(self, sessionId, messageId):
          super(Message, self).__init__()
          self.seesionId = sessionId
          self.messageId = messageId
  
  
  def pull_all():
      while stop == False:
  
          m = sub.pull(max_messages = max_messages, return_immediately = False)
  
          for data in m:
              ack_id = data[0]
              message = data[1]
              messageId = message.message_id
              data = message.data
              event = json.loads(data)
              sessionId = str(event["sessionId"])
              messages.append(Message(sessionId = sessionId, messageId = messageId))
  
              print '200 **** sessionId: ' + sessionId + ", messageId:" + messageId + " ******"
  
              sub.acknowledge(ack_ids = [ack_id])
  
  pull_all()
  </code></pre>
  
  <p>For generating session_id, sending request &amp; logging response from API:</p>
  
  <pre class="lang-js prettyprint-override"><code>// generate trackable sessionId
  var sessionId = 0
  
  var increment_session_id = function () {
    sessionId++;
    return sessionId;
  }
  
  var generate_data = function () {
    var data = {};
    // data.sessionId = faker.random.uuid();
    data.sessionId = increment_session_id();
    data.user = get_rand(userList);
    data.device = get_rand(deviceList);
    data.visitTime = new Date;
    data.location = get_rand(locationList);
    data.content = get_rand(contentList);
  
    return data;
  }
  
  var sendData = function (url, payload) {
    var request = $.ajax({
      url: url,
      contentType: 'application/json',
      method: 'POST',
      data: JSON.stringify(payload),
      error: function (xhr, status, errorThrown) {
        console.log(xhr, status, errorThrown);
        $('.result').prepend("&lt;pre id='json'&gt;" + JSON.stringify(xhr, null, 2) + "&lt;/pre&gt;")
        $('.result').prepend("&lt;div&gt;errorThrown: " + errorThrown + "&lt;/div&gt;")
        $('.result').prepend("&lt;div&gt;======FAIL=======&lt;/div&gt;&lt;div&gt;status: " + status + "&lt;/div&gt;")
      }
    }).done(function (xhr) {
      console.log(xhr);
      $('.result').prepend("&lt;div&gt;======SUCCESS=======&lt;/div&gt;&lt;pre id='json'&gt;" + JSON.stringify(payload, null, 2) + "&lt;/pre&gt;")
    })
  }
  
  $(submit_button).click(function () {
    var request_num = get_request_num();
    var request_url = get_url();
    for (var i = 0; i &lt; request_num; i++) {
      var data = generate_data();
      var loadData = changeVerb(data, 'load');
      sendData(request_url, loadData);
    }
  }) 
  </code></pre>
  
  <p><strong>UPDATE</strong></p>
  
  <p>I made a change on the API, and the issue seems to go away. The changes I made was instead of using one <code>pubsub.Client()</code> for all request, I initialized a client for every single request coming in. The new API looks like:</p>
  
  <pre class="lang-python prettyprint-override"><code>from flask import Flask, request, jsonify, render_template
  from flask_cors import CORS, cross_origin
  import simplejson as json
  from google.cloud import pubsub
  from functools import wraps
  import re
  import json
  
  
  app = Flask(__name__)
  
  ...
  
  @app.route('/publish', methods=['POST'])
  @cross_origin()
  @json_validator
  def publish_test_topic():
  
      ps = pubsub.Client()
  
  
      pubsub_topic = 'test_topic'
      data = request.data
  
      topic = ps.topic(pubsub_topic)
  
      event = json.loads(data)
  
      messageId = topic.publish(data)
      return '200 **** sessionId: ' + str(event["sessionId"]) + ", messageId:" + messageId + " ******"
  </code></pre>