Merge pull request #340 from levzem/perf-fix

CC-6187: Cache already existing index in JestElasticsearchClient

Author: kkonstantine

src/main/java/io/confluent/connect/elasticsearch/jest/JestElasticsearchClient.java

@@ -195,7 +195,11 @@ private boolean indexExists(String index) {
     Action action = new IndicesExists.Builder(index).build();
     try {
       JestResult result = client.execute(action);
-      return result.isSucceeded();
+      boolean exists = result.isSucceeded();
+      if (exists) {
+        indexCache.add(index);
+      }
+      return exists;
     } catch (IOException e) {
       throw new ConnectException(e);
     }

src/test/java/io/confluent/connect/elasticsearch/jest/JestElasticsearchClientTest.java

@@ -36,6 +36,7 @@
 import io.searchbox.indices.IndicesExists;
 import io.searchbox.indices.mapping.GetMapping;
 import io.searchbox.indices.mapping.PutMapping;
+import java.util.Collections;
 import org.apache.kafka.connect.data.Schema;
 import org.apache.kafka.connect.errors.ConnectException;
 import org.junit.Before;
@@ -56,6 +57,7 @@
 import static org.mockito.Mockito.inOrder;
 import static org.mockito.Mockito.mock;
 import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.verifyNoMoreInteractions;
 import static org.mockito.Mockito.when;
 
 public class JestElasticsearchClientTest {
@@ -89,6 +91,23 @@ public void getsVersion() {
     assertThat(client.getVersion(), is(equalTo(ElasticsearchClient.Version.ES_V1)));
   }
 
+  @Test
+  public void attemptToCreateExistingIndex() throws Exception {
+    JestElasticsearchClient client = new JestElasticsearchClient(jestClient);
+    JestResult success = new JestResult(new Gson());
+    success.setSucceeded(true);
+    IndicesExists indicesExists = new IndicesExists.Builder(INDEX).build();
+    when(jestClient.execute(indicesExists)).thenReturn(success);
+    when(jestClient.execute(argThat(isCreateIndexForTestIndex()))).thenReturn(success);
+
+    client.createIndices(Collections.singleton(INDEX));
+    InOrder inOrder = inOrder(jestClient);
+    inOrder.verify(jestClient).execute(info);
+    inOrder.verify(jestClient).execute(indicesExists);
+
+    verifyNoMoreInteractions(jestClient);
+  }
+
   @Test
   public void createsIndices() throws Exception {
     JestElasticsearchClient client = new JestElasticsearchClient(jestClient);