Propagate async init errors to token filter

Author: Mikep86

server/src/main/java/org/elasticsearch/index/analysis/StopRemoteWordListFilter.java

@@ -13,24 +13,32 @@
 import org.apache.lucene.analysis.TokenStream;
 import org.apache.lucene.analysis.tokenattributes.CharTermAttribute;
 import org.apache.lucene.util.SetOnce;
+import org.elasticsearch.ElasticsearchStatusException;
+import org.elasticsearch.rest.RestStatus;
 
 import java.io.IOException;
+import java.util.concurrent.atomic.AtomicReference;
 
 public class StopRemoteWordListFilter extends FilteringTokenFilter {
     private final SetOnce<CharArraySet> stopWordsHolder;
+    private final AtomicReference<Exception> asyncInitException;
     private final CharTermAttribute termAtt = addAttribute(CharTermAttribute.class);
 
-    public StopRemoteWordListFilter(TokenStream in, SetOnce<CharArraySet> stopWordsHolder) {
+    public StopRemoteWordListFilter(TokenStream in, SetOnce<CharArraySet> stopWordsHolder, AtomicReference<Exception> asyncInitException) {
         super(in);
         this.stopWordsHolder = stopWordsHolder;
+        this.asyncInitException = asyncInitException;
     }
 
     @Override
     protected boolean accept() throws IOException {
         // TODO: Would be nice if we could wrap a StopFilter instance and call its accept method, but it has protected access
         CharArraySet stopWords = stopWordsHolder.get();
+        Exception asyncInitException = this.asyncInitException.get();
         if (stopWords == null) {
-            throw new IllegalStateException("Stop filter not initialized yet");
+            throw asyncInitException != null
+                ? new ElasticsearchStatusException("Stop filter async initialization failed", RestStatus.INTERNAL_SERVER_ERROR, asyncInitException)
+                : new ElasticsearchStatusException("Stop filter not initialized yet", RestStatus.CONFLICT);
         }
 
         return stopWords.contains(termAtt.buffer(), 0, termAtt.length()) == false;

server/src/main/java/org/elasticsearch/index/analysis/StopRemoteWordListTokenFilterFactory.java

@@ -18,12 +18,14 @@
 
 import java.util.List;
 import java.util.Set;
+import java.util.concurrent.atomic.AtomicReference;
 
 import static org.apache.lucene.analysis.StopFilter.makeStopSet;
 
 public class StopRemoteWordListTokenFilterFactory extends AbstractTokenFilterFactory {
 
     private final SetOnce<CharArraySet> stopWordsHolder = new SetOnce<>();
+    private final AtomicReference<Exception> asyncInitException = new AtomicReference<>();
 
     private final boolean ignoreCase;
 
@@ -56,15 +58,15 @@ public void onResponse(List<String> wordList) {
 
                 @Override
                 public void onFailure(Exception e) {
-                    // TODO: Propagate error to token filter
+                    asyncInitException.set(e);
                 }
             }
         );
     }
 
     @Override
     public TokenStream create(TokenStream tokenStream) {
-        return new StopRemoteWordListFilter(tokenStream, stopWordsHolder);
+        return new StopRemoteWordListFilter(tokenStream, stopWordsHolder, asyncInitException);
     }
 
     public Set<?> stopWords() {