ElasticsearchEventSink

The event search Sink of one aggregate type: writes each commit the sink subscription hands it as one stream document (ElasticsearchEventDocuments) of the type's events index (ElasticsearchNames.events, ElasticsearchEventIndex), which the events queries of an Elasticsearch store read.

• Batch: one _bulk request per write, every record a create of its document.

• Idempotent by id: a document's id is the commit's identity (ElasticsearchNames.streamId: tenant, aggregate id and version), and a record written before is answered 409 by its create, which counts as written: a record written again changes nothing, whether a retry, a replay or an old owner of the partition writes it.

• Durable on return: Elasticsearch answers a bulk request once each document is in its shard's translog. A write does not wait for the next refresh (searches see it within the index's refresh interval, 1 s by default), unless the store's refresh is true (an immediate refresh, for tests): waiting for one would hold each aggregate's next commit for up to the interval.

• Failure: a request that fails, or any record it did not write, fails the write: transient answers (429, 502, 503, 504, a connection failure) as a StorageUnavailableException the runtime retries and classifies systemic, any other as an IllegalStateException. The records it wrote stay; the retry's creates of them answer 409.

• Access attributes: a document holds its record's attributes with the version they are of (StateEventRecord.accessVersion). After a write, each record whose attributes differ from those of the aggregate's stream before it (the record before it in the write, the last one this Sink wrote of the aggregate, capacity of them remembered, or else that stream's document) stamps them over the aggregate's earlier documents (stamp), the last such record of each aggregate only, after a refresh so the documents just written are among them. Records are written in log order, so the Sink follows the history it writes as the log does (me.ahoo.wow.spi.access.AccessStamps): a record read before a stamp of the log and one read after it differ, and the later version wins.

The index is created before the first write as the store's DDL mode says (auto creates it, validate requires it, none assumes it), and its primary waited for, as the snapshot indices are.