Use OpType.CREATE in GeoIpDownloader (elastic#69951)

When indexing new chunks in GeoIpDwonloader we should never have to overwrite old chunks, if we try to that means that there are 2 simultaneous executions. This change forces one of them to throw error in such case.

modules/ingest-geoip/src/main/java/org/elasticsearch/ingest/geoip/GeoIpDownloader.java

@@ -170,6 +170,7 @@ int indexChunks(String name, InputStream is, int chunk, String expectedMd5) thro
         for (byte[] buf = getChunk(is); buf.length != 0; buf = getChunk(is)) {
             md.update(buf);
             client.prepareIndex(DATABASES_INDEX, "_doc").setId(name + "_" + chunk)
+                .setCreate(true)
                 .setSource(XContentType.SMILE, "name", name, "chunk", chunk, "data", buf)
                 .setRefreshPolicy(WriteRequest.RefreshPolicy.IMMEDIATE)
                 .setWaitForActiveShards(ActiveShardCount.ALL)

modules/ingest-geoip/src/test/java/org/elasticsearch/ingest/geoip/GeoIpDownloaderTests.java

@@ -12,6 +12,7 @@
 import org.elasticsearch.action.ActionRequest;
 import org.elasticsearch.action.ActionResponse;
 import org.elasticsearch.action.ActionType;
+import org.elasticsearch.action.DocWriteRequest.OpType;
 import org.elasticsearch.action.index.IndexAction;
 import org.elasticsearch.action.index.IndexRequest;
 import org.elasticsearch.action.index.IndexResponse;
@@ -164,6 +165,7 @@ public void testIndexChunks() throws IOException {
 
         client.addHandler(IndexAction.INSTANCE, (IndexRequest request, ActionListener<IndexResponse> listener) -> {
             int chunk = chunkIndex.getAndIncrement();
+            assertEquals(OpType.CREATE, request.opType());
             assertEquals("test_" + (chunk + 15), request.id());
             assertEquals(XContentType.SMILE, request.getContentType());
             Map<String, Object> source = request.sourceAsMap();