[FLINK-30162] Remove unused stuff
diff --git a/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java b/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
index 9b4f9fb..8061d04 100644
--- a/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
+++ b/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchApiCallBridge.java
@@ -27,7 +27,6 @@
import java.io.IOException;
import java.io.Serializable;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
/**
@@ -49,10 +48,9 @@
/**
* Creates an Elasticsearch client implementing {@link AutoCloseable}.
*
- * @param clientConfig The configuration to use when constructing the client.
* @return The created client.
*/
- C createClient(Map<String, String> clientConfig);
+ C createClient();
/**
* Creates a {@link BulkProcessor.Builder} for creating the bulk processor.
diff --git a/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java b/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
index c0fad6f..51c5859 100644
--- a/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
+++ b/flink-connector-elasticsearch-base/src/main/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBase.java
@@ -142,15 +142,6 @@
// ------------------------------------------------------------------------
/**
- * The config map that contains configuration for the bulk flushing behaviours.
- *
- * <p>For {@link org.elasticsearch.client.transport.TransportClient} based implementations, this
- * config map would also contain Elasticsearch-shipped configuration, and therefore this config
- * map would also be forwarded when creating the Elasticsearch client.
- */
- private final Map<String, String> userConfig;
-
- /**
* The function that is used to construct multiple {@link ActionRequest ActionRequests} from
* each incoming element.
*/
@@ -296,8 +287,6 @@
} else {
bulkProcessorFlushBackoffPolicy = null;
}
-
- this.userConfig = userConfig;
}
/**
@@ -313,7 +302,7 @@
@Override
public void open(Configuration parameters) throws Exception {
- client = callBridge.createClient(userConfig);
+ client = callBridge.createClient();
callBridge.verifyClientConnection(client);
bulkProcessor = buildBulkProcessor(new BulkProcessorListener());
requestIndexer =
diff --git a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java
index 87480c5..25cb18b 100644
--- a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java
+++ b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkBaseTest.java
@@ -593,7 +593,7 @@
private static final long serialVersionUID = -4272760730959041699L;
@Override
- public Client createClient(Map<String, String> clientConfig) {
+ public Client createClient() {
return mock(Client.class);
}
diff --git a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkTestBase.java b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkTestBase.java
index 5c3d4ec..71a1e40 100644
--- a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkTestBase.java
+++ b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/ElasticsearchSinkTestBase.java
@@ -27,9 +27,7 @@
import org.elasticsearch.client.RestHighLevelClient;
import java.util.Collections;
-import java.util.HashMap;
import java.util.List;
-import java.util.Map;
import java.util.function.Function;
import static org.assertj.core.api.Assertions.assertThatThrownBy;
@@ -46,32 +44,18 @@
protected abstract RestHighLevelClient getClient();
- protected abstract String getClusterName();
-
/** Tests that the Elasticsearch sink works properly with json. */
public void runElasticsearchSinkTest() throws Exception {
runElasticSearchSinkTest(
"elasticsearch-sink-test-json-index", SourceSinkDataTestKit::getJsonSinkFunction);
}
- /** Tests that the Elasticsearch sink works properly with cbor. */
- public void runElasticsearchSinkCborTest() throws Exception {
- runElasticSearchSinkTest(
- "elasticsearch-sink-test-cbor-index", SourceSinkDataTestKit::getCborSinkFunction);
- }
-
/** Tests that the Elasticsearch sink works properly with smile. */
public void runElasticsearchSinkSmileTest() throws Exception {
runElasticSearchSinkTest(
"elasticsearch-sink-test-smile-index", SourceSinkDataTestKit::getSmileSinkFunction);
}
- /** Tests that the Elasticsearch sink works properly with yaml. */
- public void runElasticsearchSinkYamlTest() throws Exception {
- runElasticSearchSinkTest(
- "elasticsearch-sink-test-yaml-index", SourceSinkDataTestKit::getYamlSinkFunction);
- }
-
private void runElasticSearchSinkTest(
String index,
Function<String, ElasticsearchSinkFunction<Tuple2<Integer, String>>> functionFactory)
@@ -81,9 +65,7 @@
DataStreamSource<Tuple2<Integer, String>> source =
env.addSource(new SourceSinkDataTestKit.TestDataSourceFunction());
- source.addSink(
- createElasticsearchSinkForEmbeddedNode(
- 1, getClusterName(), functionFactory.apply(index)));
+ source.addSink(createElasticsearchSinkForEmbeddedNode(1, functionFactory.apply(index)));
env.execute("Elasticsearch Sink Test");
@@ -103,10 +85,7 @@
assertThatThrownBy(
() ->
createElasticsearchSink(
- 1,
- getClusterName(),
- null,
- SourceSinkDataTestKit.getJsonSinkFunction("test")))
+ 1, null, SourceSinkDataTestKit.getJsonSinkFunction("test")))
.isInstanceOfAny(IllegalArgumentException.class, NullPointerException.class);
}
@@ -118,7 +97,6 @@
() ->
createElasticsearchSink(
1,
- getClusterName(),
Collections.emptyList(),
SourceSinkDataTestKit.getJsonSinkFunction("test")))
.isInstanceOf(IllegalArgumentException.class);
@@ -134,7 +112,6 @@
source.addSink(
createElasticsearchSinkForNode(
1,
- "invalid-cluster-name",
SourceSinkDataTestKit.getJsonSinkFunction("test"),
"123.123.123.123")); // incorrect ip address
@@ -142,22 +119,9 @@
.isInstanceOf(JobExecutionException.class);
}
- /** Utility method to create a user config map. */
- protected Map<String, String> createUserConfig(int bulkFlushMaxActions, String clusterName) {
- Map<String, String> userConfig = new HashMap<>();
- userConfig.put("cluster.name", clusterName);
- userConfig.put(
- ElasticsearchSinkBase.CONFIG_KEY_BULK_FLUSH_MAX_ACTIONS,
- String.valueOf(bulkFlushMaxActions));
- userConfig.put("transport.tcp.connect_timeout", "5s");
-
- return userConfig;
- }
-
/** Creates a version-specific Elasticsearch sink, using arbitrary transport addresses. */
protected abstract ElasticsearchSinkBase<Tuple2<Integer, String>, C> createElasticsearchSink(
int bulkFlushMaxActions,
- String clusterName,
List<A> addresses,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction);
@@ -166,14 +130,12 @@
* node.
*
* <p>This case is singled out from {@link
- * ElasticsearchSinkTestBase#createElasticsearchSink(int, String, List,
- * ElasticsearchSinkFunction)} because the Elasticsearch Java API to do so is incompatible
- * across different versions.
+ * ElasticsearchSinkTestBase#createElasticsearchSink(int, List, ElasticsearchSinkFunction)}
+ * because the Elasticsearch Java API to do so is incompatible across different versions.
*/
protected abstract ElasticsearchSinkBase<Tuple2<Integer, String>, C>
createElasticsearchSinkForEmbeddedNode(
int bulkFlushMaxActions,
- String clusterName,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction)
throws Exception;
@@ -183,7 +145,6 @@
protected abstract ElasticsearchSinkBase<Tuple2<Integer, String>, C>
createElasticsearchSinkForNode(
int bulkFlushMaxActions,
- String clusterName,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction,
String ipAddress)
throws Exception;
diff --git a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/EmbeddedElasticsearchNodeEnvironment.java b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/EmbeddedElasticsearchNodeEnvironment.java
deleted file mode 100644
index 738ddd6..0000000
--- a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/EmbeddedElasticsearchNodeEnvironment.java
+++ /dev/null
@@ -1,54 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF 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.apache.flink.streaming.connectors.elasticsearch;
-
-import org.elasticsearch.client.Client;
-
-import java.io.File;
-
-/**
- * The {@link EmbeddedElasticsearchNodeEnvironment} is used in integration tests to manage
- * Elasticsearch embedded nodes.
- *
- * <p>NOTE: In order for {@link ElasticsearchSinkTestBase} to dynamically load version-specific
- * implementations for the tests, concrete implementations must be named {@code
- * EmbeddedElasticsearchNodeEnvironmentImpl}. It must also be located under the same package. The
- * intentional package-private accessibility of this interface enforces that.
- */
-public interface EmbeddedElasticsearchNodeEnvironment {
-
- /**
- * Start an embedded Elasticsearch node instance. Calling this method multiple times
- * consecutively should not restart the embedded node.
- *
- * @param tmpDataFolder The temporary data folder for the embedded node to use.
- * @param clusterName The name of the cluster that the embedded node should be configured with.
- */
- void start(File tmpDataFolder, String clusterName) throws Exception;
-
- /** Close the embedded node, if previously started. */
- void close() throws Exception;
-
- /**
- * Returns a client to communicate with the embedded node.
- *
- * @return Client to communicate with the embedded node. Returns {@code null} if the embedded
- * node wasn't started or is closed.
- */
- Client getClient();
-}
diff --git a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/ElasticsearchResource.java b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/ElasticsearchResource.java
deleted file mode 100644
index f92bcf1..0000000
--- a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/ElasticsearchResource.java
+++ /dev/null
@@ -1,106 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF 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.apache.flink.streaming.connectors.elasticsearch.testutils;
-
-import org.apache.flink.streaming.connectors.elasticsearch.EmbeddedElasticsearchNodeEnvironment;
-import org.apache.flink.util.InstantiationUtil;
-
-import org.elasticsearch.action.ActionFuture;
-import org.elasticsearch.action.admin.cluster.health.ClusterHealthRequestBuilder;
-import org.elasticsearch.action.admin.cluster.health.ClusterHealthResponse;
-import org.elasticsearch.client.AdminClient;
-import org.elasticsearch.client.Client;
-import org.elasticsearch.client.ClusterAdminClient;
-import org.elasticsearch.common.unit.TimeValue;
-import org.junit.rules.ExternalResource;
-import org.junit.rules.TemporaryFolder;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import static org.assertj.core.api.Assertions.assertThat;
-
-/** A resource that starts an embedded elasticsearch cluster. */
-public class ElasticsearchResource extends ExternalResource {
- private static final Logger LOG = LoggerFactory.getLogger(ElasticsearchResource.class);
- private EmbeddedElasticsearchNodeEnvironment embeddedNodeEnv;
- private final TemporaryFolder tempFolder = new TemporaryFolder();
-
- private final String clusterName;
-
- public ElasticsearchResource(String clusterName) {
- this.clusterName = clusterName;
- }
-
- @Override
- protected void before() throws Throwable {
-
- LOG.info("-------------------------------------------------------------------------");
- LOG.info(" Starting embedded Elasticsearch node ");
- LOG.info("-------------------------------------------------------------------------");
-
- // dynamically load version-specific implementation of the Elasticsearch embedded node
- // environment
- Class<?> clazz =
- Class.forName(
- "org.apache.flink.streaming.connectors.elasticsearch.EmbeddedElasticsearchNodeEnvironmentImpl");
- embeddedNodeEnv =
- (EmbeddedElasticsearchNodeEnvironment) InstantiationUtil.instantiate(clazz);
-
- tempFolder.create();
- embeddedNodeEnv.start(tempFolder.newFolder(), clusterName);
-
- waitForCluster();
- }
-
- /** Blocks until the cluster is ready and data nodes/nodes are live. */
- private void waitForCluster() {
- AdminClient adminClient = embeddedNodeEnv.getClient().admin();
- ClusterAdminClient clusterAdminClient = adminClient.cluster();
-
- ClusterHealthRequestBuilder requestBuilder = clusterAdminClient.prepareHealth("_all");
- requestBuilder = requestBuilder.setTimeout(TimeValue.timeValueSeconds(120));
-
- ActionFuture<ClusterHealthResponse> healthFuture =
- clusterAdminClient.health(requestBuilder.request());
-
- ClusterHealthResponse health = healthFuture.actionGet(TimeValue.timeValueSeconds(120));
-
- assertThat(health.getNumberOfNodes()).isGreaterThanOrEqualTo(1);
- assertThat(health.getNumberOfDataNodes()).isGreaterThanOrEqualTo(1);
- }
-
- @Override
- protected void after() {
-
- LOG.info("-------------------------------------------------------------------------");
- LOG.info(" Shutting down embedded Elasticsearch node ");
- LOG.info("-------------------------------------------------------------------------");
-
- try {
- embeddedNodeEnv.close();
- tempFolder.delete();
- } catch (Exception e) {
- throw new RuntimeException(e);
- }
- }
-
- public Client getClient() {
- return embeddedNodeEnv.getClient();
- }
-}
diff --git a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/SourceSinkDataTestKit.java b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/SourceSinkDataTestKit.java
index 33023a7..c286c90 100644
--- a/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/SourceSinkDataTestKit.java
+++ b/flink-connector-elasticsearch-base/src/test/java/org/apache/flink/streaming/connectors/elasticsearch/testutils/SourceSinkDataTestKit.java
@@ -74,11 +74,6 @@
}
}
- public static ElasticsearchSinkFunction<Tuple2<Integer, String>> getCborSinkFunction(
- String index) {
- return new TestElasticsearchSinkFunction(index, XContentFactory::cborBuilder);
- }
-
public static ElasticsearchSinkFunction<Tuple2<Integer, String>> getJsonSinkFunction(
String index) {
return new TestElasticsearchSinkFunction(index, XContentFactory::jsonBuilder);
@@ -89,11 +84,6 @@
return new TestElasticsearchSinkFunction(index, XContentFactory::smileBuilder);
}
- public static ElasticsearchSinkFunction<Tuple2<Integer, String>> getYamlSinkFunction(
- String index) {
- return new TestElasticsearchSinkFunction(index, XContentFactory::yamlBuilder);
- }
-
private static class TestElasticsearchSinkFunction
implements ElasticsearchSinkFunction<Tuple2<Integer, String>> {
private static final long serialVersionUID = 1L;
diff --git a/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch6/Elasticsearch6ApiCallBridge.java b/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch6/Elasticsearch6ApiCallBridge.java
index bd5d5a6..c923db3 100644
--- a/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch6/Elasticsearch6ApiCallBridge.java
+++ b/flink-connector-elasticsearch6/src/main/java/org/apache/flink/streaming/connectors/elasticsearch6/Elasticsearch6ApiCallBridge.java
@@ -38,7 +38,6 @@
import java.io.IOException;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
/** Implementation of {@link ElasticsearchApiCallBridge} for Elasticsearch 6 and later versions. */
@@ -63,7 +62,7 @@
}
@Override
- public RestHighLevelClient createClient(Map<String, String> clientConfig) {
+ public RestHighLevelClient createClient() {
RestClientBuilder builder =
RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()]));
restClientFactory.configureRestClientBuilder(builder);
diff --git a/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch6/ElasticsearchSinkITCase.java b/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch6/ElasticsearchSinkITCase.java
index 3115481..68bb99e 100644
--- a/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch6/ElasticsearchSinkITCase.java
+++ b/flink-connector-elasticsearch6/src/test/java/org/apache/flink/streaming/connectors/elasticsearch6/ElasticsearchSinkITCase.java
@@ -49,11 +49,6 @@
DockerImageVersions.ELASTICSEARCH_6, LOG);
@Override
- protected String getClusterName() {
- return "docker-cluster";
- }
-
- @Override
protected final RestHighLevelClient getClient() {
return new RestHighLevelClient(
RestClient.builder(HttpHost.create(elasticsearchContainer.getHttpHostAddress())));
@@ -88,7 +83,6 @@
protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient>
createElasticsearchSink(
int bulkFlushMaxActions,
- String clusterName,
List<HttpHost> httpHosts,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction) {
@@ -103,12 +97,10 @@
protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient>
createElasticsearchSinkForEmbeddedNode(
int bulkFlushMaxActions,
- String clusterName,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction) {
return createElasticsearchSinkForNode(
bulkFlushMaxActions,
- clusterName,
elasticsearchSinkFunction,
elasticsearchContainer.getHttpHostAddress());
}
@@ -117,7 +109,6 @@
protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient>
createElasticsearchSinkForNode(
int bulkFlushMaxActions,
- String clusterName,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction,
String hostAddress) {
diff --git a/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java b/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java
index b076a30..37a2d59 100644
--- a/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java
+++ b/flink-connector-elasticsearch7/src/main/java/org/apache/flink/streaming/connectors/elasticsearch7/Elasticsearch7ApiCallBridge.java
@@ -39,7 +39,6 @@
import java.io.IOException;
import java.util.List;
-import java.util.Map;
import java.util.concurrent.atomic.AtomicLong;
/** Implementation of {@link ElasticsearchApiCallBridge} for Elasticsearch 7 and later versions. */
@@ -64,7 +63,7 @@
}
@Override
- public RestHighLevelClient createClient(Map<String, String> clientConfig) {
+ public RestHighLevelClient createClient() {
RestClientBuilder builder =
RestClient.builder(httpHosts.toArray(new HttpHost[httpHosts.size()]));
restClientFactory.configureRestClientBuilder(builder);
diff --git a/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java b/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java
index 026d7d9..55ff8ed 100644
--- a/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java
+++ b/flink-connector-elasticsearch7/src/test/java/org/apache/flink/streaming/connectors/elasticsearch7/ElasticsearchSinkITCase.java
@@ -49,11 +49,6 @@
DockerImageVersions.ELASTICSEARCH_7, LOG);
@Override
- protected String getClusterName() {
- return "docker-cluster";
- }
-
- @Override
protected final RestHighLevelClient getClient() {
return new RestHighLevelClient(
RestClient.builder(HttpHost.create(elasticsearchContainer.getHttpHostAddress())));
@@ -88,7 +83,6 @@
protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient>
createElasticsearchSink(
int bulkFlushMaxActions,
- String clusterName,
List<HttpHost> httpHosts,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction) {
@@ -103,12 +97,10 @@
protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient>
createElasticsearchSinkForEmbeddedNode(
int bulkFlushMaxActions,
- String clusterName,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction) {
return createElasticsearchSinkForNode(
bulkFlushMaxActions,
- clusterName,
elasticsearchSinkFunction,
elasticsearchContainer.getHttpHostAddress());
}
@@ -117,7 +109,6 @@
protected ElasticsearchSinkBase<Tuple2<Integer, String>, RestHighLevelClient>
createElasticsearchSinkForNode(
int bulkFlushMaxActions,
- String clusterName,
ElasticsearchSinkFunction<Tuple2<Integer, String>> elasticsearchSinkFunction,
String hostAddress) {
