allow custom datastream type (#804)

Author: srishti-saraswat

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

@@ -378,7 +378,8 @@ public class ElasticsearchSinkConnectorConfig extends AbstractConfig {
           + "The default is %s which indicates the connector will write "
           + "to regular indices instead. If set, this configuration will "
           + "be used alongside %s and %s to construct the data stream name in the form of "
-          + "{``%s``}-{``%s``}-{``%s``}.",
+          + "{``%s``}-{``%s``}-{``%s``}. Custom index templates defined in the destination"
+          + " cluster are supported.",
       DataStreamType.NONE.name(),
       DATA_STREAM_DATASET_CONFIG,
       DATA_STREAM_NAMESPACE_CONFIG,
@@ -885,7 +886,6 @@ private static void addDataStreamConfigs(ConfigDef configDef) {
             DATA_STREAM_TYPE_CONFIG,
             Type.STRING,
             DATA_STREAM_TYPE_DEFAULT.name(),
-            new EnumRecommender<>(DataStreamType.class),
             Importance.LOW,
             DATA_STREAM_TYPE_DOC,
             DATA_STREAM_GROUP,
@@ -921,7 +921,8 @@ public boolean isBasicProxyConfigured() {
   }
 
   public boolean isDataStream() {
-    return dataStreamType() != DataStreamType.NONE && !dataStreamDataset().isEmpty();
+    return !dataStreamType().toUpperCase().equals(DataStreamType.NONE.name())
+            && !dataStreamDataset().isEmpty();
   }
 
   public boolean isProxyWithAuthenticationConfigured() {
@@ -1002,8 +1003,8 @@ public String dataStreamNamespace() {
     return getString(DATA_STREAM_NAMESPACE_CONFIG);
   }
 
-  public DataStreamType dataStreamType() {
-    return DataStreamType.valueOf(getString(DATA_STREAM_TYPE_CONFIG).toUpperCase());
+  public String dataStreamType() {
+    return getString(DATA_STREAM_TYPE_CONFIG);
   }
 
   public List<String> dataStreamTimestampField() {

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

@@ -199,7 +199,7 @@ private String convertTopicToDataStreamName(String topic) {
     }
     String dataStream = String.format(
         "%s-%s-%s",
-        config.dataStreamType().name().toLowerCase(),
+        config.dataStreamType().toLowerCase(),
         config.dataStreamDataset(),
         namespace
     );

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

@@ -138,7 +138,8 @@ private void validateCredentials() {
   }
 
   private void validateDataStreamConfigs() {
-    if (config.dataStreamType() == DataStreamType.NONE ^ config.dataStreamDataset().isEmpty()) {
+    if (config.dataStreamType().toUpperCase().equals(DataStreamType.NONE.name())
+            ^ config.dataStreamDataset().isEmpty()) {
       String errorMessage = String.format(
           "Either both or neither '%s' and '%s' must be set.",
           DATA_STREAM_DATASET_CONFIG,

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

@@ -99,8 +99,8 @@ public void shouldNotAllowInvalidCharactersDataStreamDataset() {
     new ElasticsearchSinkConnectorConfig(props);
   }
 
-  @Test(expected = ConfigException.class)
-  public void shouldNotAllowInvalidDataStreamType() {
+  @Test
+  public void shouldAllowCustomDataStreamType() {
     props.put(DATA_STREAM_TYPE_CONFIG, "notLogOrMetrics");
     new ElasticsearchSinkConnectorConfig(props);
   }