start working on indexers

modules/elasticsearch/src/main/java/org/elasticsearch/index/CloseableIndexComponent.java

@@ -17,7 +17,7 @@
  * under the License.
  */
 
-package org.elasticsearch.common.component;
+package org.elasticsearch.index;
 
 import org.elasticsearch.ElasticSearchException;
 

modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexGateway.java

@@ -19,7 +19,7 @@
 
 package org.elasticsearch.index.gateway;
 
-import org.elasticsearch.common.component.CloseableIndexComponent;
+import org.elasticsearch.index.CloseableIndexComponent;
 import org.elasticsearch.index.IndexComponent;
 
 /**

modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGateway.java

@@ -20,7 +20,7 @@
 package org.elasticsearch.index.gateway;
 
 import org.elasticsearch.ElasticSearchIllegalStateException;
-import org.elasticsearch.common.component.CloseableIndexComponent;
+import org.elasticsearch.index.CloseableIndexComponent;
 import org.elasticsearch.index.deletionpolicy.SnapshotIndexCommit;
 import org.elasticsearch.index.shard.IndexShardComponent;
 import org.elasticsearch.index.translog.Translog;

modules/elasticsearch/src/main/java/org/elasticsearch/index/gateway/IndexShardGatewayService.java

@@ -19,11 +19,11 @@
 
 package org.elasticsearch.index.gateway;
 
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.unit.ByteSizeValue;
 import org.elasticsearch.common.unit.TimeValue;
+import org.elasticsearch.index.CloseableIndexComponent;
 import org.elasticsearch.index.deletionpolicy.SnapshotIndexCommit;
 import org.elasticsearch.index.engine.Engine;
 import org.elasticsearch.index.engine.EngineException;

modules/elasticsearch/src/main/java/org/elasticsearch/index/service/IndexService.java

@@ -21,8 +21,8 @@ package org.elasticsearch.index.service;
 
 import org.elasticsearch.ElasticSearchException;
 import org.elasticsearch.common.collect.ImmutableSet;
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.inject.Injector;
+import org.elasticsearch.index.CloseableIndexComponent;
 import org.elasticsearch.index.IndexComponent;
 import org.elasticsearch.index.IndexShardMissingException;
 import org.elasticsearch.index.cache.IndexCache;

modules/elasticsearch/src/main/java/org/elasticsearch/index/service/InternalIndexService.java

@@ -24,17 +24,13 @@ import org.elasticsearch.ElasticSearchInterruptedException;
 import org.elasticsearch.common.collect.ImmutableMap;
 import org.elasticsearch.common.collect.ImmutableSet;
 import org.elasticsearch.common.collect.UnmodifiableIterator;
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.inject.Injector;
 import org.elasticsearch.common.inject.Injectors;
 import org.elasticsearch.common.inject.ModulesBuilder;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.gateway.none.NoneGateway;
-import org.elasticsearch.index.AbstractIndexComponent;
-import org.elasticsearch.index.Index;
-import org.elasticsearch.index.IndexShardAlreadyExistsException;
-import org.elasticsearch.index.IndexShardMissingException;
+import org.elasticsearch.index.*;
 import org.elasticsearch.index.cache.IndexCache;
 import org.elasticsearch.index.deletionpolicy.DeletionPolicyModule;
 import org.elasticsearch.index.engine.Engine;

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/CloseableIndexerComponent.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+import org.elasticsearch.ElasticSearchException;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public interface CloseableIndexerComponent {
+
+    /**
+     * Closes the indexer component. A boolean indicating if its part of an actual index
+     * deletion or not is passed.
+     *
+     * @param delete <tt>true</tt> if the index is being deleted.
+     * @throws org.elasticsearch.ElasticSearchException
+     *
+     */
+    void close(boolean delete) throws ElasticSearchException;
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/Indexer.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public interface Indexer extends IndexerComponent {
+
+    void close(boolean delete);
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexerComponent.java

@@ -0,0 +1,28 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public interface IndexerComponent {
+
+    IndexerName indexerName();
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexerException.java

@@ -0,0 +1,47 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+import org.elasticsearch.ElasticSearchException;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexerException extends ElasticSearchException {
+
+    private final IndexerName indexer;
+
+    public IndexerException(IndexerName indexer, String msg) {
+        this(indexer, msg, null);
+    }
+
+    public IndexerException(IndexerName indexer, String msg, Throwable cause) {
+        this(indexer, true, msg, cause);
+    }
+
+    protected IndexerException(IndexerName indexer, boolean withSpace, String msg, Throwable cause) {
+        super("[" + indexer.type() + "][" + indexer.name() + "]" + (withSpace ? " " : "") + msg, cause);
+        this.indexer = indexer;
+    }
+
+    public IndexerName indexerName() {
+        return indexer;
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexerModule.java

@@ -0,0 +1,83 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+import org.elasticsearch.common.Strings;
+import org.elasticsearch.common.collect.ImmutableList;
+import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.inject.Module;
+import org.elasticsearch.common.inject.Modules;
+import org.elasticsearch.common.inject.SpawnModules;
+import org.elasticsearch.common.settings.NoClassSettingsException;
+import org.elasticsearch.common.settings.Settings;
+
+import static org.elasticsearch.common.Strings.*;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexerModule extends AbstractModule implements SpawnModules {
+
+    private IndexerName indexerName;
+
+    private final Settings settings;
+
+    public IndexerModule(IndexerName indexerName, Settings settings) {
+        this.indexerName = indexerName;
+        this.settings = settings;
+    }
+
+    @Override public Iterable<? extends Module> spawnModules() {
+        String type = settings.get("indexer.type");
+        if (type == null) {
+            return ImmutableList.of(Modules.createModule(loadTypeModule(indexerName.type(), "org.elasticsearch.indexer.", "IndexerModule"), settings));
+        } else {
+            return ImmutableList.of(Modules.createModule(settings.getAsClass("indexer.type", Module.class, "org.elasticsearch.indexer.", "IndexerModule"), settings));
+        }
+    }
+
+    @Override protected void configure() {
+
+    }
+
+    private Class<? extends Module> loadTypeModule(String type, String prefixPackage, String suffixClassName) {
+        String fullClassName = type;
+        try {
+            return (Class<? extends Module>) settings.getClassLoader().loadClass(fullClassName);
+        } catch (ClassNotFoundException e) {
+            fullClassName = prefixPackage + Strings.capitalize(toCamelCase(type)) + suffixClassName;
+            try {
+                return (Class<? extends Module>) settings.getClassLoader().loadClass(fullClassName);
+            } catch (ClassNotFoundException e1) {
+                fullClassName = prefixPackage + toCamelCase(type) + "." + Strings.capitalize(toCamelCase(type)) + suffixClassName;
+                try {
+                    return (Class<? extends Module>) settings.getClassLoader().loadClass(fullClassName);
+                } catch (ClassNotFoundException e2) {
+                    fullClassName = prefixPackage + toCamelCase(type).toLowerCase() + "." + Strings.capitalize(toCamelCase(type)) + suffixClassName;
+                    try {
+                        return (Class<? extends Module>) settings.getClassLoader().loadClass(fullClassName);
+                    } catch (ClassNotFoundException e3) {
+                        throw new NoClassSettingsException("Failed to load class with value [" + type + "]", e);
+                    }
+                }
+            }
+        }
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexerName.java

@@ -0,0 +1,69 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexerName {
+
+    private final String type;
+
+    private final String name;
+
+    public IndexerName(String type, String name) {
+        this.type = type;
+        this.name = name;
+    }
+
+    public String type() {
+        return this.type;
+    }
+
+    public String getType() {
+        return type();
+    }
+
+    public String name() {
+        return this.name;
+    }
+
+    public String getName() {
+        return name();
+    }
+
+    @Override public boolean equals(Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        IndexerName that = (IndexerName) o;
+
+        if (name != null ? !name.equals(that.name) : that.name != null) return false;
+        if (type != null ? !type.equals(that.type) : that.type != null) return false;
+
+        return true;
+    }
+
+    @Override public int hashCode() {
+        int result = type != null ? type.hashCode() : 0;
+        result = 31 * result + (name != null ? name.hashCode() : 0);
+        return result;
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexerNameModule.java

@@ -0,0 +1,38 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+import org.elasticsearch.common.inject.AbstractModule;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexerNameModule extends AbstractModule {
+
+    private final IndexerName indexerName;
+
+    public IndexerNameModule(IndexerName indexerName) {
+        this.indexerName = indexerName;
+    }
+
+    @Override protected void configure() {
+        bind(IndexerName.class).toInstance(indexerName);
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexersModule.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.settings.Settings;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexersModule extends AbstractModule {
+
+    private final Settings settings;
+
+    public IndexersModule(Settings settings) {
+        this.settings = settings;
+    }
+
+    @Override protected void configure() {
+        bind(IndexersService.class).asEagerSingleton();
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/IndexersService.java

@@ -0,0 +1,154 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer;
+
+import org.elasticsearch.ElasticSearchException;
+import org.elasticsearch.common.collect.ImmutableMap;
+import org.elasticsearch.common.collect.ImmutableSet;
+import org.elasticsearch.common.collect.MapBuilder;
+import org.elasticsearch.common.collect.Maps;
+import org.elasticsearch.common.component.AbstractLifecycleComponent;
+import org.elasticsearch.common.inject.Inject;
+import org.elasticsearch.common.inject.Injector;
+import org.elasticsearch.common.inject.Injectors;
+import org.elasticsearch.common.inject.ModulesBuilder;
+import org.elasticsearch.common.settings.Settings;
+import org.elasticsearch.indexer.settings.IndexerSettingsModule;
+import org.elasticsearch.threadpool.ThreadPool;
+
+import java.util.Map;
+import java.util.concurrent.CountDownLatch;
+
+import static org.elasticsearch.common.settings.ImmutableSettings.*;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexersService extends AbstractLifecycleComponent<IndexersService> {
+
+    private final ThreadPool threadPool;
+
+    private final Injector injector;
+
+    private final Map<IndexerName, Injector> indexersInjectors = Maps.newHashMap();
+
+    private volatile ImmutableMap<IndexerName, Indexer> indexers = ImmutableMap.of();
+
+    @Inject public IndexersService(Settings settings, ThreadPool threadPool, Injector injector) {
+        super(settings);
+        this.threadPool = threadPool;
+        this.injector = injector;
+    }
+
+    @Override protected void doStart() throws ElasticSearchException {
+    }
+
+    @Override protected void doStop() throws ElasticSearchException {
+        ImmutableSet<IndexerName> indices = ImmutableSet.copyOf(this.indexers.keySet());
+        final CountDownLatch latch = new CountDownLatch(indices.size());
+        for (final IndexerName indexerName : indices) {
+            threadPool.cached().execute(new Runnable() {
+                @Override public void run() {
+                    try {
+                        deleteIndexer(indexerName, false);
+                    } catch (Exception e) {
+                        logger.warn("failed to delete indexer on stop [{}]/[{}]", e, indexerName.type(), indexerName.name());
+                    } finally {
+                        latch.countDown();
+                    }
+                }
+            });
+        }
+        try {
+            latch.await();
+        } catch (InterruptedException e) {
+            // ignore
+        }
+    }
+
+    @Override protected void doClose() throws ElasticSearchException {
+    }
+
+    public synchronized Indexer createIndexer(IndexerName indexerName, Settings settings) throws ElasticSearchException {
+        if (indexersInjectors.containsKey(indexerName)) {
+            throw new IndexerException(indexerName, "indexer already exists");
+        }
+
+        logger.debug("creating indexer [{}][{}]", indexerName.type(), indexerName.name());
+
+        Settings indexerSettings = settingsBuilder()
+                .put(this.settings)
+                .put(settings)
+                .classLoader(settings.getClassLoader())
+                .globalSettings(settings.getGlobalSettings())
+                .build();
+
+        ModulesBuilder modules = new ModulesBuilder();
+        modules.add(new IndexerNameModule(indexerName));
+        modules.add(new IndexerSettingsModule(indexerSettings));
+        modules.add(new IndexerModule(indexerName, indexerSettings));
+
+        Injector indexInjector = modules.createChildInjector(injector);
+        indexersInjectors.put(indexerName, indexInjector);
+        Indexer indexer = indexInjector.getInstance(Indexer.class);
+        indexers = MapBuilder.newMapBuilder(indexers).put(indexerName, indexer).immutableMap();
+        return indexer;
+    }
+
+    public synchronized void cleanIndexer(IndexerName indexerName) throws ElasticSearchException {
+        deleteIndexer(indexerName, false);
+    }
+
+    public synchronized void deleteIndexer(IndexerName indexerName) throws ElasticSearchException {
+        deleteIndexer(indexerName, true);
+    }
+
+    private void deleteIndexer(IndexerName indexerName, boolean delete) {
+        Injector indexerInjector;
+        Indexer indexer;
+        synchronized (this) {
+            indexerInjector = indexersInjectors.remove(indexerName);
+            if (indexerInjector == null) {
+                if (!delete) {
+                    return;
+                }
+                throw new IndexerException(indexerName, "missing");
+            }
+            if (delete) {
+                logger.debug("deleting indexer [{}][{}]", indexerName.type(), indexerName.name());
+            }
+
+            Map<IndexerName, Indexer> tmpMap = Maps.newHashMap(indexers);
+            indexer = tmpMap.remove(indexerName);
+            indexers = ImmutableMap.copyOf(tmpMap);
+        }
+
+//        for (Class<? extends CloseableIndexerComponent> closeable : pluginsService.indexServices()) {
+//            indexerInjector.getInstance(closeable).close(delete);
+//        }
+
+        indexer.close(delete);
+
+        indexerInjector.getInstance(Indexer.class).close(delete);
+
+        Injectors.close(injector);
+
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/settings/IndexerSettings.java

@@ -0,0 +1,40 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer.settings;
+
+import org.elasticsearch.common.inject.BindingAnnotation;
+
+import java.lang.annotation.Documented;
+import java.lang.annotation.Retention;
+import java.lang.annotation.Target;
+
+import static java.lang.annotation.ElementType.*;
+import static java.lang.annotation.RetentionPolicy.*;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+
+@BindingAnnotation
+@Target({FIELD, PARAMETER})
+@Retention(RUNTIME)
+@Documented
+public @interface IndexerSettings {
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indexer/settings/IndexerSettingsModule.java

@@ -0,0 +1,39 @@
+/*
+ * Licensed to Elastic Search and Shay Banon under one
+ * or more contributor license agreements.  See the NOTICE file
+ * distributed with this work for additional information
+ * regarding copyright ownership. Elastic Search licenses this
+ * file to you under the Apache License, Version 2.0 (the
+ * "License"); you may not use this file except in compliance
+ * with the License.  You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the License is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+ * KIND, either express or implied.  See the License for the
+ * specific language governing permissions and limitations
+ * under the License.
+ */
+
+package org.elasticsearch.indexer.settings;
+
+import org.elasticsearch.common.inject.AbstractModule;
+import org.elasticsearch.common.settings.Settings;
+
+/**
+ * @author kimchy (shay.banon)
+ */
+public class IndexerSettingsModule extends AbstractModule {
+
+    private final Settings settings;
+
+    public IndexerSettingsModule(Settings settings) {
+        this.settings = settings;
+    }
+
+    @Override protected void configure() {
+        bind(Settings.class).annotatedWith(IndexerSettings.class).toInstance(settings);
+    }
+}

modules/elasticsearch/src/main/java/org/elasticsearch/indices/InternalIndicesService.java

@@ -29,7 +29,6 @@ import org.elasticsearch.common.collect.ImmutableMap;
 import org.elasticsearch.common.collect.ImmutableSet;
 import org.elasticsearch.common.collect.UnmodifiableIterator;
 import org.elasticsearch.common.component.AbstractLifecycleComponent;
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.inject.Injector;
 import org.elasticsearch.common.inject.Injectors;

modules/elasticsearch/src/main/java/org/elasticsearch/node/internal/InternalNode.java

@@ -54,6 +54,8 @@ import org.elasticsearch.gateway.GatewayModule;
 import org.elasticsearch.gateway.GatewayService;
 import org.elasticsearch.http.HttpServer;
 import org.elasticsearch.http.HttpServerModule;
+import org.elasticsearch.indexer.IndexersModule;
+import org.elasticsearch.indexer.IndexersService;
 import org.elasticsearch.indices.IndicesModule;
 import org.elasticsearch.indices.IndicesService;
 import org.elasticsearch.indices.cluster.IndicesClusterStateService;
@@ -132,6 +134,7 @@ public final class InternalNode implements Node {
         if (settings.getAsBoolean("http.enabled", true)) {
             modules.add(new HttpServerModule(settings));
         }
+        modules.add(new IndexersModule(settings));
         modules.add(new IndicesModule(settings));
         modules.add(new SearchModule());
         modules.add(new TransportActionModule());
@@ -168,6 +171,7 @@ public final class InternalNode implements Node {
 
         injector.getInstance(IndicesService.class).start();
         injector.getInstance(IndicesClusterStateService.class).start();
+        injector.getInstance(IndexersService.class).start();
         injector.getInstance(ClusterService.class).start();
         injector.getInstance(RoutingService.class).start();
         injector.getInstance(SearchService.class).start();
@@ -205,6 +209,7 @@ public final class InternalNode implements Node {
         injector.getInstance(MonitorService.class).stop();
         injector.getInstance(GatewayService.class).stop();
         injector.getInstance(SearchService.class).stop();
+        injector.getInstance(IndexersService.class).stop();
         injector.getInstance(IndicesClusterStateService.class).stop();
         injector.getInstance(IndicesService.class).stop();
         injector.getInstance(RestController.class).stop();
@@ -250,6 +255,8 @@ public final class InternalNode implements Node {
         injector.getInstance(GatewayService.class).close();
         stopWatch.stop().start("search");
         injector.getInstance(SearchService.class).close();
+        stopWatch.stop().start("indexers");
+        injector.getInstance(IndexersService.class).close();
         stopWatch.stop().start("indices_cluster");
         injector.getInstance(IndicesClusterStateService.class).close();
         stopWatch.stop().start("indices");

modules/elasticsearch/src/main/java/org/elasticsearch/plugins/AbstractPlugin.java

@@ -20,9 +20,9 @@
 package org.elasticsearch.plugins;
 
 import org.elasticsearch.common.collect.ImmutableList;
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.component.LifecycleComponent;
 import org.elasticsearch.common.inject.Module;
+import org.elasticsearch.index.CloseableIndexComponent;
 
 import java.util.Collection;
 

modules/elasticsearch/src/main/java/org/elasticsearch/plugins/Plugin.java

@@ -19,9 +19,9 @@
 
 package org.elasticsearch.plugins;
 
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.component.LifecycleComponent;
 import org.elasticsearch.common.inject.Module;
+import org.elasticsearch.index.CloseableIndexComponent;
 
 import java.util.Collection;
 

modules/elasticsearch/src/main/java/org/elasticsearch/plugins/PluginsService.java

@@ -24,13 +24,13 @@ import org.elasticsearch.common.collect.ImmutableMap;
 import org.elasticsearch.common.collect.Lists;
 import org.elasticsearch.common.collect.Maps;
 import org.elasticsearch.common.component.AbstractComponent;
-import org.elasticsearch.common.component.CloseableIndexComponent;
 import org.elasticsearch.common.component.LifecycleComponent;
 import org.elasticsearch.common.inject.Inject;
 import org.elasticsearch.common.inject.Module;
 import org.elasticsearch.common.io.Streams;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.env.Environment;
+import org.elasticsearch.index.CloseableIndexComponent;
 
 import java.io.*;
 import java.lang.reflect.Method;