The `DatabaseNodeService` is responsible for synchronizing and updating geoIP data on ingest nodes that is used by the GeoIP processor to perform lookups on incoming data.

This race condition is somewhat low risk. The impacts are that the update process of geoIP data can fail and will need to be retried after the next cluster state update. If the update process fails, a small window exists on the starting node where incoming geoIP data will be marked with the `DatabaseUnavailableProcessor` and need to be reprocessed. Example log of the issue is as follows:

```
[2022-05-17T21:36:32,599][ERROR][o.e.i.g.DatabaseNodeService] [elasticsearch2] failed to retrieve database [GeoLite2-City.mmdb]
org.elasticsearch.cluster.block.ClusterBlockException: blocked by: [SERVICE_UNAVAILABLE/1/state not recovered / initialized];
	at org.elasticsearch.cluster.block.ClusterBlocks.globalBlockedException(ClusterBlocks.java:177) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.cluster.block.ClusterBlocks.globalBlockedRaiseException(ClusterBlocks.java:163) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.search.TransportSearchAction.executeSearch(TransportSearchAction.java:913) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.search.TransportSearchAction.executeLocalSearch(TransportSearchAction.java:747) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.search.TransportSearchAction.lambda$executeRequest$6(TransportSearchAction.java:390) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.ActionListener$1.onResponse(ActionListener.java:136) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.index.query.Rewriteable.rewriteAndFetch(Rewriteable.java:112) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.index.query.Rewriteable.rewriteAndFetch(Rewriteable.java:77) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.search.TransportSearchAction.executeRequest(TransportSearchAction.java:478) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.search.TransportSearchAction.doExecute(TransportSearchAction.java:277) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.search.TransportSearchAction.doExecute(TransportSearchAction.java:103) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.support.TransportAction$RequestFilterChain.proceed(TransportAction.java:79) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.support.ActionFilter$Simple.apply(ActionFilter.java:53) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.support.TransportAction$RequestFilterChain.proceed(TransportAction.java:77) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.action.support.TransportAction.execute(TransportAction.java:54) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.tasks.TaskManager.registerAndExecute(TaskManager.java:170) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.node.NodeClient.executeLocally(NodeClient.java:105) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.node.NodeClient.doExecute(NodeClient.java:83) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.support.AbstractClient.execute(AbstractClient.java:380) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.FilterClient.doExecute(FilterClient.java:57) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.OriginSettingClient.doExecute(OriginSettingClient.java:43) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.support.AbstractClient.execute(AbstractClient.java:380) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.support.AbstractClient.execute(AbstractClient.java:366) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.client.internal.support.AbstractClient.search(AbstractClient.java:510) ~[elasticsearch-8.1.3.jar:8.1.3]
	at org.elasticsearch.ingest.geoip.DatabaseNodeService.lambda$retrieveDatabase$13(DatabaseNodeService.java:395) [ingest-geoip-8.1.3.jar:8.1.3]
	at org.elasticsearch.common.util.concurrent.ThreadContext$ContextPreservingRunnable.run(ThreadContext.java:717) [elasticsearch-8.1.3.jar:8.1.3]
	at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1136) [?:?]
	at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:635) [?:?]
```

The root cause of this race condition is that `DatabaseNodeService` receives updates to the cluster state by registering itself with the `IngestService#addIngestClusterStateListener`. The `IngestService` is a `ClusterStateApplier` so the cluster state received during updates **has not yet been published** to the `ClusterService` as the current cluster state. When a node is starting, the `IngestService` will see that this state is the first one that lacks the "cluster state not recovered" block on it and will start the geoIP database update process. The update process forks to another thread which eventually calls the search action to pull the database contents locally out of the system index that holds them.

If the forked update process gets to the search operation before the cluster state can be fully committed by the previous thread, AND if the client that submits the search operation deems it appropriate to execute the search against a locally available shard, the search operation will read the current cluster state (which still has the not "recovered block" on it) and throw a `ClusterBlockException`. This causes the download process to fail on the forked thread. The next cluster state update that arrives will restart the process, which should not encounter this race condition again since the preconditions for execution (no cluster blocks) should be stable at that point.
modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpProcessorFactoryTests.java
GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseNodeService, clusterService);
        Map<String, Object> config = new HashMap<>();
        config.put("field", "source_field");
        config.put("fallback_to_default_databases", randomBoolean());
        factory.create(null, null, null, config);
        assertWarnings(GeoIpProcessor.DEFAULT_DATABASES_DEPRECATION_MESSAGE);
    }
    public void testDefaultDatabaseWithTaskPresent() throws Exception {
        PersistentTasksCustomMetadata tasks = PersistentTasksCustomMetadata.builder()
            .addTask(GeoIpDownloader.GEOIP_DOWNLOADER, GeoIpDownloader.GEOIP_DOWNLOADER, null, null)
            .updateTaskState(GeoIpDownloader.GEOIP_DOWNLOADER, GeoIpTaskState.EMPTY)
            .build();
        ClusterState clusterState = ClusterState.builder(ClusterState.EMPTY_STATE)
            .metadata(Metadata.builder().putCustom(PersistentTasksCustomMetadata.TYPE, tasks))
            .build();
        when(clusterService.state()).thenReturn(clusterState);
        GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseNodeService, clusterService);
        Map<String, Object> config = new HashMap<>();
        config.put("field", "_field");
        String processorTag = randomAlphaOfLength(10);
        GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, processorTag, null, config);
        processor.execute(RandomDocumentPicks.randomIngestDocument(random(), Map.of("_field", "89.160.20.128")));
    }
    public void testUpdateDatabaseWhileIngesting() throws Exception {
        GeoIpProcessor.Factory factory = new GeoIpProcessor.Factory(databaseNodeService, clusterService);
        Map<String, Object> config = new HashMap<>();
        config.put("field", "source_field");
        GeoIpProcessor processor = (GeoIpProcessor) factory.create(null, null, null, config);
        Map<String, Object> document = new HashMap<>();
        document.put("source_field", "89.160.20.128");