iron-core/src/main/java/io/axway/iron/core/internal/StorePersistence.java
Optional<BigInteger> loadStores(Function<String, EntityStores> entityStoresByStoreName,
                                    BiFunction<SerializableSnapshot, String, SerializableSnapshot> snapshotPostProcessor) {
        Optional<BigInteger> latestSnapshotTxId;
        try {
            latestSnapshotTxId = m_snapshotStore.listSnapshots().stream().max(BigInteger::compareTo);
        } catch (IOException e) {
            throw new UnrecoverableStoreException("Error occurred when recovering from latest snapshot", e);
        }

        latestSnapshotTxId.ifPresent(lastTx -> {
            LOG.info("Recovering store from snapshot", args -> args.add("transactionId", lastTx));
            var postProcess = new SnapshotPostProcessor(snapshotPostProcessor);
            try {
                Flowable.fromPublisher(m_snapshotStore.createSnapshotReader(lastTx))  //
                        .blockingForEach(reader -> {
                            String storeName = reader.storeName();
                            EntityStores entityStores = entityStoresByStoreName.apply(storeName);

                            SerializableSnapshot serializableSnapshot;
                            try (InputStream is = reader.inputStream()) {
                                serializableSnapshot = m_snapshotSerializer.deserializeSnapshot(storeName, is);
                            }
                            if (serializableSnapshot.getSnapshotModelVersion() != SNAPSHOT_MODEL_VERSION) {
                                throw new UnrecoverableStoreException("Snapshot serializable model version is not supported",
                                                                      args -> args.add("version", serializableSnapshot.getSnapshotModelVersion())
                                                                              .add("expectedVersion", SNAPSHOT_MODEL_VERSION));
                            }

                            if (!lastTx.equals(serializableSnapshot.getTransactionId())) {
                                throw new UnrecoverableStoreException("Snapshot transaction id  mismatch with request transaction id",
                                                                      args -> args.add("snapshotTransactionId", serializableSnapshot.getTransactionId())
                                                                              .add("requestTransactionId", lastTx));
                            }

                            SerializableSnapshot finalSnapshot = postProcess.apply(storeName, serializableSnapshot);

                            finalSnapshot.getEntities().forEach(serializableEntityInstances -> {
                                String entityName = serializableEntityInstances.getEntityName();
                                EntityStore<?> entityStore = entityStores.getEntityStore(entityName);
                                checkArgument(entityStore != null, "Entity has not be registered in the store", args -> args.add("entityName", entityName));

                                entityStore.recover(serializableEntityInstances);
                            });
                        });
            } catch (Exception e) {
                throw new UnrecoverableStoreException("Error occurred when recovering from latest snapshot", e);
            }

            // update the applicationModelVersion if any consistent load/update
            m_applicationModelVersion = postProcess.getConsistentApplicationModelVersion();
        });

        if (!latestSnapshotTxId.isPresent()) {
            LOG.info("Store has no snapshot, store is empty, creating it's first snapshot");
        }

        return latestSnapshotTxId;
    }
Optional<BigInteger> loadStores(Function<String, EntityStores> entityStoresByStoreName,
                                    BiFunction<SerializableSnapshot, String, SerializableSnapshot> snapshotPostProcessor) {
        Optional<BigInteger> latestSnapshotTxId;
        try {
            latestSnapshotTxId = m_snapshotStore.listSnapshots().stream().max(BigInteger::compareTo);
        } catch (IOException e) {
            throw new UnrecoverableStoreException("Error occurred when recovering from latest snapshot", e);
        }

        latestSnapshotTxId.ifPresent(lastTx -> {
            LOG.info("Recovering store from snapshot", args -> args.add("transactionId", lastTx));
            var postProcess = new SnapshotPostProcessor(snapshotPostProcessor);
            try {
                Flowable.fromPublisher(m_snapshotStore.createSnapshotReader(lastTx))  //
                        .blockingForEach(reader -> {
                            String storeName = reader.storeName();
                            EntityStores entityStores = entityStoresByStoreName.apply(storeName);

                            SerializableSnapshot serializableSnapshot;
                            try (InputStream is = reader.inputStream()) {
                                serializableSnapshot = m_snapshotSerializer.deserializeSnapshot(storeName, is);
                            }
                            if (serializableSnapshot.getSnapshotModelVersion() != SNAPSHOT_MODEL_VERSION) {
                                throw new UnrecoverableStoreException("Snapshot serializable model version is not supported",
                                                                      args -> args.add("version", serializableSnapshot.getSnapshotModelVersion())
                                                                              .add("expectedVersion", SNAPSHOT_MODEL_VERSION));
                            }

                            if (!lastTx.equals(serializableSnapshot.getTransactionId())) {
                                throw new UnrecoverableStoreException("Snapshot transaction id  mismatch with request transaction id",
                                                                      args -> args.add("snapshotTransactionId", serializableSnapshot.getTransactionId())
                                                                              .add("requestTransactionId", lastTx));
                            }

                            SerializableSnapshot finalSnapshot = postProcess.apply(storeName, serializableSnapshot);

                            finalSnapshot.getEntities().forEach(serializableEntityInstances -> {
                                String entityName = serializableEntityInstances.getEntityName();
                                EntityStore<?> entityStore = entityStores.getEntityStore(entityName);
                                checkArgument(entityStore != null, "Entity has not be registered in the store", args -> args.add("entityName", entityName));

                                entityStore.recover(serializableEntityInstances);
                            });
                        });
            } catch (Exception e) {
                throw new UnrecoverableStoreException("Error occurred when recovering from latest snapshot", e);
            }

            // update the applicationModelVersion if any consistent load/update
            m_applicationModelVersion = postProcess.getConsistentApplicationModelVersion();
        });

        if (!latestSnapshotTxId.isPresent()) {
            LOG.info("Store has no snapshot, store is empty, creating it's first snapshot");
        }

        return latestSnapshotTxId;
    }
Load the stores.

@param entityStoresByStoreName the map store name to entity stores
@return latestSnapshotTxId the transaction id of the last snapshot