CC-12530: do not throw exception on flush after task close (#455)

* CC-12530: do not throw exception on flush after task close

Signed-off-by: Lev Zemlyanov <lev@confluent.io>

Author: Lev Zemlyanov

src/main/java/io/confluent/connect/elasticsearch/DataConverter.java

@@ -323,7 +323,7 @@ private Object preProcessLogicalValue(String schemaName, Object value) {
   }
 
   private Object preProcessArrayValue(Object value, Schema schema, Schema newSchema) {
-    Collection collection = (Collection) value;
+    Collection<?> collection = (Collection<?>) value;
     List<Object> result = new ArrayList<>();
     for (Object element: collection) {
       result.add(preProcessValue(element, schema.valueSchema(), newSchema.valueSchema()));

src/main/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTask.java

@@ -35,7 +35,7 @@
 public class ElasticsearchSinkTask extends SinkTask {
 
   private static final Logger log = LoggerFactory.getLogger(ElasticsearchSinkTask.class);
-  private ElasticsearchWriter writer;
+  private volatile ElasticsearchWriter writer;
   private ElasticsearchClient client;
   private Boolean createIndicesAtStartTime;
 
@@ -124,8 +124,12 @@ public void put(Collection<SinkRecord> records) throws ConnectException {
 
   @Override
   public void flush(Map<TopicPartition, OffsetAndMetadata> offsets) {
-    log.trace("Flushing data to Elasticsearch with the following offsets: {}", offsets);
-    writer.flush();
+    if (writer != null) {
+      log.debug("Flushing data to Elasticsearch with the following offsets: {}", offsets);
+      writer.flush();
+    } else {
+      log.debug("Could not flush data to Elasticsearch because ESWriter already closed.");
+    }
   }
 
   @Override
@@ -138,6 +142,7 @@ public void stop() throws ConnectException {
     log.info("Stopping ElasticsearchSinkTask.");
     if (writer != null) {
       writer.stop();
+      writer = null;
     }
     if (client != null) {
       client.close();

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

@@ -316,7 +316,7 @@ public BulkRequest createBulkRequest(List<IndexableRecord> batch) {
   }
 
   // visible for testing
-  protected BulkableAction toBulkableAction(IndexableRecord record) {
+  protected BulkableAction<?> toBulkableAction(IndexableRecord record) {
     // If payload is null, the record was a tombstone and we should delete from the index.
     return record.payload != null ? toIndexRequest(record) : toDeleteRequest(record);
   }

src/test/java/io/confluent/connect/elasticsearch/DataConverterTest.java

@@ -235,7 +235,7 @@ public void stringKeyedMapCompactFormat() {
         SchemaBuilder.map(Schema.STRING_SCHEMA, Schema.INT32_SCHEMA).build(),
         preProcessedSchema
     );
-    HashMap newValue = (HashMap) converter.preProcessValue(origValue, origSchema, preProcessedSchema);
+    HashMap<?, ?> newValue = (HashMap<?, ?>) converter.preProcessValue(origValue, origSchema, preProcessedSchema);
     assertEquals(origValue, newValue);
   }
 

src/test/java/io/confluent/connect/elasticsearch/ElasticsearchSinkTaskTest.java

@@ -152,6 +152,14 @@ public void testCreateAndWriteToIndexNotCreatedAtStartTime() {
 
   }
 
+  @Test
+  public void testStopThenFlushDoesNotThrow() {
+    ElasticsearchSinkTask task = new ElasticsearchSinkTask();
+    task.start(createProps(), client);
+    task.stop();
+    task.flush(new HashMap<>());
+  }
+
   private boolean verifyIndexExist(InternalTestCluster cluster, String ... indices) {
     ActionFuture<IndicesExistsResponse> action = cluster
         .client()

src/test/java/io/confluent/connect/elasticsearch/bulk/BulkProcessorTest.java

@@ -379,10 +379,8 @@ public void ignoreOrWarnOnMalformedDoc() throws InterruptedException {
     // Test both IGNORE and WARN options
     // There is no difference in logic between IGNORE and WARN, except for the logging.
     // Test to ensure they both work the same logically
-    final List<BehaviorOnMalformedDoc> behaviorsToTest = new ArrayList<BehaviorOnMalformedDoc>() {{
-      add(BehaviorOnMalformedDoc.WARN);
-      add(BehaviorOnMalformedDoc.IGNORE);
-    }};
+    final List<BehaviorOnMalformedDoc> behaviorsToTest =
+        Arrays.asList(BehaviorOnMalformedDoc.WARN, BehaviorOnMalformedDoc.IGNORE);
 
     for(BehaviorOnMalformedDoc behaviorOnMalformedDoc : behaviorsToTest)
     {