From ded5b773daeb14de3e42cdd0c67e3c36aff1e232 Mon Sep 17 00:00:00 2001
From: Shay Banon <kimchy@gmail.com>
Date: Sat, 23 Jun 2012 20:26:27 +0200
Subject: [PATCH] River name re-use not possible between node shutdowns, closes
 #1921. This effectively happens because we don't flush before deleting a
 mapping, which causes it to still reply the changes happening for it on the
 transaction log, meaning the mapping are still around on the index level.

---
 .../delete/TransportDeleteMappingAction.java  | 32 ++++++++++++++-----
 1 file changed, 24 insertions(+), 8 deletions(-)

diff --git a/src/main/java/org/elasticsearch/action/admin/indices/mapping/delete/TransportDeleteMappingAction.java b/src/main/java/org/elasticsearch/action/admin/indices/mapping/delete/TransportDeleteMappingAction.java
index ec3b305c9f8..689fedb9f23 100644
--- a/src/main/java/org/elasticsearch/action/admin/indices/mapping/delete/TransportDeleteMappingAction.java
+++ b/src/main/java/org/elasticsearch/action/admin/indices/mapping/delete/TransportDeleteMappingAction.java
@@ -21,6 +21,8 @@ package org.elasticsearch.action.admin.indices.mapping.delete;
 
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.action.ActionListener;
+import org.elasticsearch.action.admin.indices.flush.FlushResponse;
+import org.elasticsearch.action.admin.indices.flush.TransportFlushAction;
 import org.elasticsearch.action.admin.indices.refresh.RefreshResponse;
 import org.elasticsearch.action.admin.indices.refresh.TransportRefreshAction;
 import org.elasticsearch.action.deletebyquery.DeleteByQueryResponse;
@@ -49,6 +51,8 @@ public class TransportDeleteMappingAction extends TransportMasterNodeOperationAc
 
     private final MetaDataMappingService metaDataMappingService;
 
+    private final TransportFlushAction flushAction;
+
     private final TransportDeleteByQueryAction deleteByQueryAction;
 
     private final TransportRefreshAction refreshAction;
@@ -56,11 +60,12 @@ public class TransportDeleteMappingAction extends TransportMasterNodeOperationAc
     @Inject
     public TransportDeleteMappingAction(Settings settings, TransportService transportService, ClusterService clusterService,
                                         ThreadPool threadPool, MetaDataMappingService metaDataMappingService,
-                                        TransportDeleteByQueryAction deleteByQueryAction, TransportRefreshAction refreshAction) {
+                                        TransportDeleteByQueryAction deleteByQueryAction, TransportRefreshAction refreshAction, TransportFlushAction flushAction) {
         super(settings, transportService, clusterService, threadPool);
         this.metaDataMappingService = metaDataMappingService;
         this.deleteByQueryAction = deleteByQueryAction;
         this.refreshAction = refreshAction;
+        this.flushAction = flushAction;
     }
 
     @Override
@@ -100,19 +105,30 @@ public class TransportDeleteMappingAction extends TransportMasterNodeOperationAc
 
         final AtomicReference<Throwable> failureRef = new AtomicReference<Throwable>();
         final CountDownLatch latch = new CountDownLatch(1);
-        deleteByQueryAction.execute(Requests.deleteByQueryRequest(request.indices()).query(QueryBuilders.filteredQuery(QueryBuilders.matchAllQuery(), FilterBuilders.typeFilter(request.type()))), new ActionListener<DeleteByQueryResponse>() {
+        flushAction.execute(Requests.flushRequest(request.indices()), new ActionListener<FlushResponse>() {
             @Override
-            public void onResponse(DeleteByQueryResponse deleteByQueryResponse) {
-                refreshAction.execute(Requests.refreshRequest(request.indices()), new ActionListener<RefreshResponse>() {
+            public void onResponse(FlushResponse flushResponse) {
+                deleteByQueryAction.execute(Requests.deleteByQueryRequest(request.indices()).query(QueryBuilders.filteredQuery(QueryBuilders.matchAllQuery(), FilterBuilders.typeFilter(request.type()))), new ActionListener<DeleteByQueryResponse>() {
                     @Override
-                    public void onResponse(RefreshResponse refreshResponse) {
-                        metaDataMappingService.removeMapping(new MetaDataMappingService.RemoveRequest(request.indices(), request.type()));
-                        latch.countDown();
+                    public void onResponse(DeleteByQueryResponse deleteByQueryResponse) {
+                        refreshAction.execute(Requests.refreshRequest(request.indices()), new ActionListener<RefreshResponse>() {
+                            @Override
+                            public void onResponse(RefreshResponse refreshResponse) {
+                                metaDataMappingService.removeMapping(new MetaDataMappingService.RemoveRequest(request.indices(), request.type()));
+                                latch.countDown();
+                            }
+
+                            @Override
+                            public void onFailure(Throwable e) {
+                                metaDataMappingService.removeMapping(new MetaDataMappingService.RemoveRequest(request.indices(), request.type()));
+                                latch.countDown();
+                            }
+                        });
                     }
 
                     @Override
                     public void onFailure(Throwable e) {
-                        metaDataMappingService.removeMapping(new MetaDataMappingService.RemoveRequest(request.indices(), request.type()));
+                        failureRef.set(e);
                         latch.countDown();
                     }
                 });