You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pulsar.apache.org by te...@apache.org on 2023/01/15 06:24:46 UTC
[pulsar] branch branch-2.11 updated: Elasticsearch sink: Support loading config from secrets (#19206)
This is an automated email from the ASF dual-hosted git repository.
technoboy pushed a commit to branch branch-2.11
in repository https://gitbox.apache.org/repos/asf/pulsar.git
The following commit(s) were added to refs/heads/branch-2.11 by this push:
new ed1a8c4f531 Elasticsearch sink: Support loading config from secrets (#19206)
ed1a8c4f531 is described below
commit ed1a8c4f531a2617a70cdc3a9c6059ebbca50c23
Author: Alexander Preuß <al...@streamnative.io>
AuthorDate: Sun Jan 15 07:24:40 2023 +0100
Elasticsearch sink: Support loading config from secrets (#19206)
---
pulsar-io/elastic-search/pom.xml | 6 ++
.../io/elasticsearch/ElasticSearchConfig.java | 7 ++-
.../pulsar/io/elasticsearch/ElasticSearchSink.java | 2 +-
.../io/elasticsearch/ElasticSearchConfigTests.java | 71 +++++++++++++++-------
4 files changed, 59 insertions(+), 27 deletions(-)
diff --git a/pulsar-io/elastic-search/pom.xml b/pulsar-io/elastic-search/pom.xml
index 7a7d035d034..1a84ae53042 100644
--- a/pulsar-io/elastic-search/pom.xml
+++ b/pulsar-io/elastic-search/pom.xml
@@ -57,6 +57,12 @@
<scope>provided</scope>
</dependency>
+ <dependency>
+ <groupId>${project.groupId}</groupId>
+ <artifactId>pulsar-io-common</artifactId>
+ <version>${project.version}</version>
+ </dependency>
+
<dependency>
<groupId>com.fasterxml.jackson.core</groupId>
<artifactId>jackson-databind</artifactId>
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
index 1dbfa0568ba..e7ea0fbaca0 100644
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfig.java
@@ -28,6 +28,8 @@ import java.util.Map;
import lombok.Data;
import lombok.experimental.Accessors;
import org.apache.commons.lang3.StringUtils;
+import org.apache.pulsar.io.common.IOConfigUtils;
+import org.apache.pulsar.io.core.SinkContext;
import org.apache.pulsar.io.core.annotations.FieldDoc;
/**
@@ -324,9 +326,8 @@ public class ElasticSearchConfig implements Serializable {
return mapper.readValue(new File(yamlFile), ElasticSearchConfig.class);
}
- public static ElasticSearchConfig load(Map<String, Object> map) throws IOException {
- ObjectMapper mapper = new ObjectMapper();
- return mapper.readValue(new ObjectMapper().writeValueAsString(map), ElasticSearchConfig.class);
+ public static ElasticSearchConfig load(Map<String, Object> map, SinkContext sinkContext) throws IOException {
+ return IOConfigUtils.loadWithSecrets(map, ElasticSearchConfig.class, sinkContext);
}
public void validate() {
diff --git a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
index 263dcc4ea6e..295cbc1da63 100644
--- a/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
+++ b/pulsar-io/elastic-search/src/main/java/org/apache/pulsar/io/elasticsearch/ElasticSearchSink.java
@@ -73,7 +73,7 @@ public class ElasticSearchSink implements Sink<GenericObject> {
@Override
public void open(Map<String, Object> config, SinkContext sinkContext) throws Exception {
- elasticSearchConfig = ElasticSearchConfig.load(config);
+ elasticSearchConfig = ElasticSearchConfig.load(config, sinkContext);
elasticSearchConfig.validate();
elasticsearchClient = new ElasticSearchClient(elasticSearchConfig);
if (!Strings.isNullOrEmpty(elasticSearchConfig.getPrimaryFields())) {
diff --git a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java
index 463a712c34c..d152b936857 100644
--- a/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java
+++ b/pulsar-io/elastic-search/src/test/java/org/apache/pulsar/io/elasticsearch/ElasticSearchConfigTests.java
@@ -18,11 +18,12 @@
*/
package org.apache.pulsar.io.elasticsearch;
+import org.apache.pulsar.io.core.SinkContext;
+import org.mockito.Mockito;
import org.testng.annotations.Test;
import java.io.File;
import java.io.IOException;
-import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
@@ -34,6 +35,8 @@ import static org.testng.Assert.expectThrows;
public class ElasticSearchConfigTests {
+ private final SinkContext mockContext = Mockito.mock(SinkContext.class);
+
@Test
public final void loadFromYamlFileTest() throws IOException {
File yamlFile = getFile("sinkConfig.yaml");
@@ -57,7 +60,7 @@ public class ElasticSearchConfigTests {
map.put("password", "go-speedie-go");
map.put("primaryFields", "x");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
assertNotNull(config);
assertEquals(config.getElasticSearchUrl(), "http://localhost:90902");
assertEquals(config.getIndexName(), "myIndex");
@@ -69,8 +72,8 @@ public class ElasticSearchConfigTests {
@Test
public final void defaultValueTest() throws IOException {
- ElasticSearchConfig config = ElasticSearchConfig.load(Collections.emptyMap());
- assertNull(config.getElasticSearchUrl());
+ Map<String, Object> requiredConfig = Map.of("elasticSearchUrl", "http://localhost:90902");
+ ElasticSearchConfig config = ElasticSearchConfig.load(requiredConfig, mockContext);
assertNull(config.getIndexName());
assertEquals(config.getTypeName(), "_doc");
assertNull(config.getUsername());
@@ -120,7 +123,7 @@ public class ElasticSearchConfigTests {
map.put("username", "racerX");
map.put("password", "go-speedie-go");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
assertNotNull(config);
config.validate();
}
@@ -134,17 +137,17 @@ public class ElasticSearchConfigTests {
map.put("password", "go-speedie-go");
map.put("indexNumberOfReplicas", "0");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@Test(expectedExceptions = IllegalArgumentException.class,
- expectedExceptionsMessageRegExp = "elasticSearchUrl not set.")
+ expectedExceptionsMessageRegExp = "elasticSearchUrl cannot be null")
public final void missingRequiredPropertiesTest() throws IOException {
Map<String, Object> map = new HashMap<String, Object> ();
map.put("indexName", "toto");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -158,7 +161,7 @@ public class ElasticSearchConfigTests {
map.put("password", "go-speedie-go");
map.put("indexNumberOfShards", "0");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -172,7 +175,7 @@ public class ElasticSearchConfigTests {
map.put("password", "go-speedie-go");
map.put("indexNumberOfReplicas", "-1");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -184,7 +187,7 @@ public class ElasticSearchConfigTests {
map.put("indexName", "myindex");
map.put("username", "racerX");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -196,7 +199,7 @@ public class ElasticSearchConfigTests {
map.put("indexName", "myindex");
map.put("password", "go-speedie-go");
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -209,35 +212,35 @@ public class ElasticSearchConfigTests {
map.put("password", "go-speedie-go");
map.put("token", "tok");
{
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
expectThrows(IllegalArgumentException.class, () -> config.validate());
}
map.put("apiKey", "apiKey");
{
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
expectThrows(IllegalArgumentException.class, () -> config.validate());
}
map.remove("token");
{
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
expectThrows(IllegalArgumentException.class, () -> config.validate());
}
map.remove("username");
map.remove("password");
{
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
map.put("token", "tok");
map.remove("apiKey");
{
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
map.remove("token");
{
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
}
@@ -248,7 +251,7 @@ public class ElasticSearchConfigTests {
Map<String, Object> map = new HashMap<String, Object> ();
map.put("elasticSearchUrl", "http://localhost:90902");
map.put("connectTimeoutInMs", -1);
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -258,7 +261,7 @@ public class ElasticSearchConfigTests {
Map<String, Object> map = new HashMap<String, Object> ();
map.put("elasticSearchUrl", "http://localhost:90902");
map.put("connectionRequestTimeoutInMs", -1);
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -268,7 +271,7 @@ public class ElasticSearchConfigTests {
Map<String, Object> map = new HashMap<String, Object> ();
map.put("elasticSearchUrl", "http://localhost:90902");
map.put("socketTimeoutInMs", -1);
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -278,7 +281,7 @@ public class ElasticSearchConfigTests {
Map<String, Object> map = new HashMap<String, Object> ();
map.put("elasticSearchUrl", "http://localhost:90902");
map.put("bulkConcurrentRequests", -1);
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
}
@@ -304,7 +307,7 @@ public class ElasticSearchConfigTests {
sslMap.put("provider", "Sun");
map.put("ssl", sslMap);
- ElasticSearchConfig config = ElasticSearchConfig.load(map);
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, mockContext);
config.validate();
assertEquals(config.getSsl().isEnabled(), true);
assertEquals(config.getSsl().getTruststorePath(), "/ssl/truststore.jks");
@@ -315,4 +318,26 @@ public class ElasticSearchConfigTests {
assertEquals(config.getSsl().getProtocols(), "TLSv1.2,TLSv1.3");
assertEquals(config.getSsl().getProvider(), "Sun");
}
+
+ @Test
+ public final void loadConfigFromSecretsTest() throws IOException {
+ SinkContext contextWithSecrets = Mockito.mock(SinkContext.class);
+ Mockito.when(contextWithSecrets.getSecret("username")).thenReturn("secretUser");
+ Mockito.when(contextWithSecrets.getSecret("password")).thenReturn("$ecret123");
+
+ Map<String, Object> map = new HashMap<String, Object> ();
+ map.put("elasticSearchUrl", "http://localhost:90902");
+ map.put("indexName", "myIndex");
+ map.put("typeName", "doc");
+ map.put("primaryFields", "x");
+
+ ElasticSearchConfig config = ElasticSearchConfig.load(map, contextWithSecrets);
+ assertNotNull(config);
+ assertEquals(config.getElasticSearchUrl(), "http://localhost:90902");
+ assertEquals(config.getIndexName(), "myIndex");
+ assertEquals(config.getTypeName(), "doc");
+ assertEquals(config.getPrimaryFields(), "x");
+ assertEquals(config.getUsername(), "secretUser");
+ assertEquals(config.getPassword(), "$ecret123");
+ }
}
\ No newline at end of file