title: Spring data MongoDB change stream with multiple application instances
body: <p>I have a springboot with   springdata  mongodb application where I am connecting to mongo change stream to save the changes to a audit collection.  My application is running multiple instances (2 instances) and will be scaled up to n number instances when the load increased.   When records are created in the original collection (“my collection”), the listeners will be triggered in all running instances and creates duplicate records.  Following is my setup</p>
  <p>build.gradle</p>
  <pre><code>…
  // spring data mingodb version 3.1.5
  implementation 'org.springframework.boot:spring-boot-starter-data-mongodb'
  …
  </code></pre>
  <p>Listener config</p>
  <pre><code>@Configuration
  @Slf4j
  public class MongoChangeStreamListenerConfig {
  
    @Bean
    MessageListenerContainer changeStreamListenerContainer(
        MongoTemplate template,
        PartyConsentAuditListener consentAuditListener,
        ErrorHandler errorHandler) {
  
      MessageListenerContainer messageListenerContainer =
          new MongoStreamListenerContainer(template, errorHandler);
  
      ChangeStreamRequest&lt;PartyConsentEntity&gt; request =
          ChangeStreamRequest.builder(consentAuditListener)
              .collection(&quot;my-collection&quot;)
              .filter(newAggregation(match(where(&quot;operationType&quot;).in(&quot;insert&quot;, &quot;update&quot;, &quot;replace&quot;))))
              .fullDocumentLookup(FullDocument.UPDATE_LOOKUP)
              .build();
      messageListenerContainer.register(request, MyEntity.class, errorHandler);
      log.info(&quot;mongo stream listener is registered&quot;);
      return messageListenerContainer;
    }
  
    @Bean
    ErrorHandler getLoggingErrorHandler() {
      return new ErrorHandler() {
        @Override
        public void handleError(Throwable throwable) {
          log.error(&quot;error in creating audit records {}&quot;, throwable);
        }
      };
    }
  }
  </code></pre>
  <p>Listener container</p>
  <pre><code>public class MongoStreamListenerContainer extends DefaultMessageListenerContainer {
  
    public MongoStreamListenerContainer(MongoTemplate template, ErrorHandler errorHandler) {
      super(template, Executors.newFixedThreadPool(15), errorHandler);
    }
  
    @Override
    public boolean isAutoStartup() {
      return true;
    }
  }
  </code></pre>
  <p>ChangeListener</p>
  <pre><code>@Component
  @Slf4j
  @RequiredArgsConstructor
  /**
   * This class will listen to mongodb change stream and process changes. The onMessage will triggered
   * when a record added, updated or replaced in the mongo db.
   */
  public class MyEntityAuditListener
      implements MessageListener&lt;ChangeStreamDocument&lt;Document&gt;, MyEntity&gt; {
  
  
    @Override
    public void onMessage(Message&lt;ChangeStreamDocument&lt;Document&gt;, MyEntity &gt; message) {
      var update = message.getBody();
      log.info(&quot;db change event received&quot;);
      if (update != null) {
        log.info(&quot;creating audit entries for id {}&quot;, update.getId());
        // This will execute in all the instances and creating duplicating records
  
      }
    }
  }
  </code></pre>
  <p>Is there a way to control the execution on one instance at a given time and share the load between nodes?.  It would be really nice to know if there is a config from spring data mongodb to control the flow.</p>
  <p>Also, I have checked the following post in stack overflow and I am not sure how to use this with spring data.
  <a href="https://stackoverflow.com/questions/52007693/mongo-change-streams-running-multiple-times-kind-of-node-app-running-multiple">Mongo Change Streams running multiple times (kind of): Node app running multiple instances</a></p>
  <p>Any help or tip to resolve this issue is highly appreciated.  Thank you very much in advance.</p>