monstache v6.7.21 and secondaryPreferred updates

[arekborucki]

We use monstache v6.7.21 and noticed that when performing a large number of write operations to MongoDB, not all update operations are replicated to Elasticsearch via monstache with secondaryPreferred read preference. However, all insert operations are replicated. Replication works fully (inserts and updates) if monstache is connected to MongoDB primary or to one of the Secondaries using the directConnection=true option and MongoDB Write URL (mongo-write-url).
We ran two types of tests: one for inserts and one for updates. Please see the results below:

We ran two types of tests: one for inserts and one for updates. Please see the results below:

Number of Operations	Number of Replicated Inserts	Number of Replicated Updates
5000	5000	4994
5000	5000	4993
5000	5000	4994
5000	5000	4995
5000	5000	4991
5000	5000	4993
5000	5000	4995
5000	5000	4998
5000	5000	4997
5000	5000	4994

We use MongoDB v6.0 , monstache v6.7.21 Monstache uses a MongoDB view as the replication source in MongoDB. Here is our configuration:

elasticsearch-urls = ["${elastic_url}"]
relate-threads = 6000
relate-buffer = 15000
elasticsearch-max-seconds = 10
elasticsearch-max-bytes = 16777216
resume = true
resume-name = "bc2-${resume_id}-contacts"
change-stream-namespaces = [ "${mongodb_database}.contacts" ]
gzip = true
stats = true
elasticsearch-retry = true
prune-invalid-json = true
dropped-databases = false
dropped-collections = false
elasticsearch-client-timeout = 30
enable-http-server = true

[[mapping]]
namespace = "${mongodb_database}.contacts"
index = "contacts${es_suffix}"

[[mapping]]
namespace = "${mongodb_database}.contacts-view"
index = "contacts${es_suffix}"

[[relate]]
namespace = "${mongodb_database}.contacts"
with-namespace = "${mongodb_database}.contacts-view"
keep-src = false

db.adminCommand({ getDefaultRWConcern: 1 })
{
  defaultReadConcern: { level: 'local' },
  defaultWriteConcern: { w: 'majority', wtimeout: 0 },
  updateOpTime: Timestamp({ t: 1717599777, i: 6 }),
  updateWallClockTime: ISODate("2024-06-05T15:02:57.764Z"),
  defaultWriteConcernSource: 'global',
  defaultReadConcernSource: 'implicit',
  localUpdateWallClockTime: ISODate("2024-06-05T15:02:57.765Z"),
  ok: 1,
  '$clusterTime': {
    clusterTime: Timestamp({ t: 1717673598, i: 4 }),
    signature: {
      hash: Binary(Buffer.from("0000000000000000000000000000000000000000", "hex"), 0),
      keyId: Long("0")
    }
  },
  operationTime: Timestamp({ t: 1717673598, i: 4 })
}