title: Google Cloud PubSub send the message to more than one consumer (in the same subscription)
body: <p>I have a Java SpringBoot2 application (app1) that sends messages to a Google Cloud PubSub topic (it is the publisher).</p>
  <p>Other Java SpringBoot2 application (app2) is subscribed to a subscription to receive those messages. But in this case, I have more than one instance (the k8s auto-scaling is enabled), so I have more than one pod for this app consuming messages from the PubSub.</p>
  <p>Some messages are consumed by one instance of app2, but many others are sent to more than one app2 instance, so the messages process is duplicated for these messages.</p>
  <p>Here is the code of consumer (app2):</p>
  <pre><code>private final static int ACK_DEAD_LINE_IN_SECONDS = 30;
  private static final long POLLING_PERIOD_MS = 250L;
  private static final int WINDOW_MAX_SIZE = 1000;
  private static final Duration WINDOW_MAX_TIME = Duration.ofSeconds(1L);
  
  @Autowired
  private PubSubAdmin pubSubAdmin;
  
  @Bean
  public ApplicationRunner runner(PubSubReactiveFactory reactiveFactory) {
      return args -&gt; {
          createSubscription(&quot;subscription-id&quot;, &quot;topic-id&quot;, ACK_DEAD_LINE_IN_SECONDS);
          reactiveFactory.poll(subscription, POLLING_PERIOD_MS) // Poll the PubSub periodically
              .map(msg -&gt; Pair.of(msg, getMessageValue(msg))) // Extract the message as a pair
              .bufferTimeout(WINDOW_MAX_SIZE, WINDOW_MAX_TIME) // Create a buffer of messages to bulk process 
              .flatMap(this::processBuffer) // Process the buffer
              .doOnError(e -&gt; log.error(&quot;Error processing event window&quot;, e))
              .retry()
              .subscribe();
      };
  }
  
  private void createSubscription(String subscriptionName, String topicName, int ackDeadline) {
      pubSubAdmin.createTopic(topicName);
      try {
          pubSubAdmin.createSubscription(subscriptionName, topicName, ackDeadline);
      } catch (AlreadyExistsException e) {
          log.info(&quot;Pubsub subscription '{}' already configured for topic '{}': {}&quot;, subscriptionName, topicName, e.getMessage());
      }
  }
  
  private Flux&lt;Void&gt; processBuffer(List&lt;Pair&lt;AcknowledgeablePubsubMessage, PreparedRecordEvent&gt;&gt; msgsWindow) {
      return Flux.fromStream(
          msgsWindow.stream()
              .collect(Collectors.groupingBy(msg -&gt; msg.getRight().getData())) // Group the messages by same data
              .values()
              .stream()
      )
      .flatMap(this::processDataBuffer);
  }
  
  private Mono&lt;Void&gt; processDataBuffer(List&lt;Pair&lt;AcknowledgeablePubsubMessage, PreparedRecordEvent&gt;&gt; dataMsgsWindow) {
      return processData(
          dataMsgsWindow.get(0).getRight().getData(),
          dataMsgsWindow.stream()
              .map(Pair::getRight)
              .map(PreparedRecordEvent::getRecord)
              .collect(Collectors.toSet())
      )
      .doOnSuccess(it -&gt;
          dataMsgsWindow.forEach(msg -&gt; {
              log.info(&quot;Mark msg ACK&quot;);
              msg.getLeft().ack();
          })
      )
      .doOnError(e -&gt; {
          log.error(&quot;Error on PreparedRecordEvent event&quot;, e);
          dataMsgsWindow.forEach(msg -&gt; {
              log.error(&quot;Mark msg NACK&quot;);
              msg.getLeft().nack();
          });
      })
      .retry();
  }
  
  private Mono&lt;Void&gt; processData(Data data, Set&lt;Record&gt; records) {
      // For each message, make calculations over the records associated to the data
      final DataQuality calculated = calculatorService.calculateDataQualityFor(data, records); // Arithmetic calculations
      return this.daasClient.updateMetrics(calculated) // Update DB record with a DaaS to wrap DB access
          .flatMap(it -&gt; {
              if (it.getProcessedRows() &gt;= it.getValidRows()) {
                  return finish(data);
              }
              return Mono.just(data);
          })
          .then();
  }
  
  private Mono&lt;Data&gt; finish(Data data) {
      return dataClient.updateStatus(data.getId, DataStatus.DONE) // Update DB record with a DaaS to wrap DB access
          .doOnSuccess(updatedData -&gt; pubSubClient.publish(
              new Qa0DonedataEvent(updatedData) // Publis a new event in other topic
          ))
          .doOnError(err -&gt; {
              log.error(&quot;Error finishing data&quot;);
          })
          .onErrorReturn(data);
  }
  </code></pre>
  <p>I need that each messages is consumed by one and only one app2 instance. Anybody know if this is possible? Any idea to achieve this?</p>
  <p>Maybe the right way is to create one subscription for each app2 instance and configure the topic to send each message t exactly one subscription instead of to every one. It is possible?</p>
  <p>According to the <a href="https://cloud.google.com/pubsub/docs/subscriber?hl=en" rel="nofollow noreferrer">official documentation</a>, once a message is sent to a subscriber, Pub/Sub tries not to deliver it to any other subscriber on the same subscription (app2 instances are subscriber of the same subscription):</p>
  <blockquote>
  <p>Once a message is sent to a subscriber, the subscriber should
  acknowledge the message. A message is considered outstanding once it
  has been sent out for delivery and before a subscriber acknowledges
  it. Pub/Sub will repeatedly attempt to deliver any message that has
  not been acknowledged. While a message is outstanding to a subscriber,
  however, Pub/Sub tries not to deliver it to any other subscriber on
  the same subscription. The subscriber has a configurable, limited
  amount of time -- known as the ackDeadline -- to acknowledge the
  outstanding message. Once the deadline passes, the message is no
  longer considered outstanding, and Pub/Sub will attempt to redeliver
  the message</p>
  </blockquote>