Reject external versioning when script is configured in OpenSearch sink (opensearch-project#6773)

OpenSearch does not support external versioning with scripted upserts.
Add config-time validation in IndexConfiguration to fail fast with a
clear error message instead of getting runtime 400 errors from OpenSearch.

Signed-off-by: Dinu John <86094133+dinujoh@users.noreply.github.com>

Author: dinujoh

data-prepper-plugins/opensearch/src/main/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexConfiguration.java

@@ -302,6 +302,13 @@ public static IndexConfiguration readIndexConfig(final OpenSearchSinkConfig open
             builder = builder.withScriptConfiguration(scriptConfiguration);
         }
 
+        if (scriptConfiguration != null && openSearchSinkConfig.getVersionType() != null
+                && VersionType.External.jsonValue().equals(openSearchSinkConfig.getVersionType())) {
+            throw new InvalidPluginConfigurationException(
+                    "document_version_type 'external' is incompatible with script configuration. " +
+                    "OpenSearch does not support external versioning with scripted upserts.");
+        }
+
         AwsAuthenticationConfiguration awsAuthenticationConfiguration = openSearchSinkConfig.getAwsAuthenticationOptions();
         if (awsAuthenticationConfiguration != null) {
             builder = builder.withServerless(awsAuthenticationConfiguration.isServerlessCollection());

data-prepper-plugins/opensearch/src/test/java/org/opensearch/dataprepper/plugins/sink/opensearch/index/IndexConfigurationTests.java

@@ -653,4 +653,26 @@ private String createTemplateContent() {
                 "\"mappings\":{\"properties\":{\"timestamp\":{\"type\":\"date\",\"format\":\"yyyy-MM-ddHH:mm:ss||yyyy-MM-dd||epoch_millis\"}," +
                 "\"value\":{\"type\":\"double\"}}}}}";
     }
+
+    @Test
+    public void testReadIndexConfig_withExternalVersionAndScript_throws_InvalidPluginConfigurationException() throws JsonProcessingException {
+        final Map<String, Object> metadata = new HashMap<>();
+        metadata.put("index_type", "custom");
+        metadata.put("index", "test-index");
+        metadata.put("document_version_type", "external");
+        metadata.put("script", Map.of("source", "ctx._source.counter += 1"));
+        final OpenSearchSinkConfig openSearchSinkConfig = getOpenSearchSinkConfig(metadata);
+        assertThrows(InvalidPluginConfigurationException.class, () -> IndexConfiguration.readIndexConfig(openSearchSinkConfig));
+    }
+
+    @Test
+    public void testReadIndexConfig_withScriptAndNoExternalVersion_doesNotThrow() throws JsonProcessingException {
+        final Map<String, Object> metadata = new HashMap<>();
+        metadata.put("index_type", "custom");
+        metadata.put("index", "test-index");
+        metadata.put("script", Map.of("source", "ctx._source.counter += 1"));
+        final OpenSearchSinkConfig openSearchSinkConfig = getOpenSearchSinkConfig(metadata);
+        final IndexConfiguration config = IndexConfiguration.readIndexConfig(openSearchSinkConfig);
+        assertThat(config.getScriptConfiguration(), notNullValue());
+    }
 }