This is an automated email from the ASF dual-hosted git repository.
zbendhiba pushed a commit to branch camel-main
in repository https://gitbox.apache.org/repos/asf/camel-quarkus.git
The following commit(s) were added to refs/heads/camel-main by this push:
new 478f3c5 System.properties not recognized if set in Test Resource -
part 2 fixes #2572
478f3c5 is described below
commit 478f3c53ba2792aef2977df49a2d5d866943d797
Author: Zineb Bendhiba <[email protected]>
AuthorDate: Tue May 11 22:04:19 2021 +0200
System.properties not recognized if set in Test Resource - part 2 fixes
#2572
---
.../wiremock/WireMockTestResourceLifecycleManager.java | 4 ++--
.../component/digitalocean/it/DigitaloceanRoute.java | 9 ++++++---
.../digitalocean/it/DigitaloceanDropletTest.java | 5 +++--
.../component/digitalocean/it/DigitaloceanTest.java | 9 +++------
.../component/geocoder/it/GeocoderProducers.java | 9 ++++++---
.../component/lumberjack/it/LumberjackTest.java | 3 ++-
.../quarkus/component/slack/it/SlackResource.java | 18 +++++++++++++-----
.../camel/quarkus/component/slack/it/SlackTest.java | 8 +++++++-
.../quarkus/component/slack/it/SlackTestResource.java | 7 +++++--
.../camel/quarkus/component/splunk/it/SplunkTest.java | 3 ++-
.../quarkus/component/twilio/it/TwilioResource.java | 14 ++++++++------
.../component/vertx/kafka/it/VertxKafkaProducers.java | 5 +++--
12 files changed, 60 insertions(+), 34 deletions(-)
diff --git
a/integration-tests-support/wiremock/src/main/java/org/apache/camel/quarkus/test/wiremock/WireMockTestResourceLifecycleManager.java
b/integration-tests-support/wiremock/src/main/java/org/apache/camel/quarkus/test/wiremock/WireMockTestResourceLifecycleManager.java
index cdf47b7..349b714 100644
---
a/integration-tests-support/wiremock/src/main/java/org/apache/camel/quarkus/test/wiremock/WireMockTestResourceLifecycleManager.java
+++
b/integration-tests-support/wiremock/src/main/java/org/apache/camel/quarkus/test/wiremock/WireMockTestResourceLifecycleManager.java
@@ -34,6 +34,7 @@ import
com.github.tomakehurst.wiremock.recording.SnapshotRecordResult;
import com.github.tomakehurst.wiremock.stubbing.StubMapping;
import io.quarkus.test.common.QuarkusTestResourceLifecycleManager;
import org.apache.camel.quarkus.test.mock.backend.MockBackendUtils;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.jboss.logging.Logger;
import static com.github.tomakehurst.wiremock.client.WireMock.recordSpec;
@@ -175,8 +176,7 @@ public abstract class WireMockTestResourceLifecycleManager
implements QuarkusTes
* Get the value of a given environment variable or a default value if it
does not exist
*/
protected String envOrDefault(String envVarName, String defaultValue) {
- String value = System.getenv(envVarName);
- return value != null ? value : defaultValue;
+ return ConfigProvider.getConfig().getOptionalValue(envVarName,
String.class).orElse(defaultValue);
}
/**
diff --git
a/integration-tests/digitalocean/src/main/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanRoute.java
b/integration-tests/digitalocean/src/main/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanRoute.java
index 243ad0a..e90a1cf 100644
---
a/integration-tests/digitalocean/src/main/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanRoute.java
+++
b/integration-tests/digitalocean/src/main/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanRoute.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.quarkus.component.digitalocean.it;
+import java.util.Optional;
+
import javax.enterprise.context.ApplicationScoped;
import javax.inject.Named;
import javax.ws.rs.Produces;
@@ -23,6 +25,7 @@ import javax.ws.rs.Produces;
import com.myjeeva.digitalocean.impl.DigitalOceanClient;
import io.quarkus.arc.Unremovable;
import org.apache.camel.builder.RouteBuilder;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.eclipse.microprofile.config.inject.ConfigProperty;
@ApplicationScoped
@@ -41,9 +44,9 @@ public class DigitaloceanRoute extends RouteBuilder {
@Unremovable
@Named("digitalOceanClient")
DigitalOceanClient initDigitalOceanClient(MockApiService mockApiService) {
- final String wireMockUrl = System.getProperty("wiremock.url.ssl");
- if (wireMockUrl != null) {
- return mockApiService.createDigitalOceanClient(wireMockUrl,
oAuthToken);
+ Optional<String> wireMockUrl =
ConfigProvider.getConfig().getOptionalValue("wiremock.url.ssl", String.class);
+ if (wireMockUrl.isPresent()) {
+ return mockApiService.createDigitalOceanClient(wireMockUrl.get(),
oAuthToken);
}
return new DigitalOceanClient(oAuthToken);
}
diff --git
a/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanDropletTest.java
b/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanDropletTest.java
index 5f00d2e..3ded48e 100644
---
a/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanDropletTest.java
+++
b/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanDropletTest.java
@@ -29,6 +29,7 @@ import io.quarkus.test.junit.QuarkusTest;
import io.restassured.RestAssured;
import io.restassured.http.ContentType;
import org.apache.camel.quarkus.test.wiremock.MockServer;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -53,8 +54,8 @@ public class DigitaloceanDropletTest {
public static void initTimeoutUnit() {
// add timeout if not using MockServer
// when using a Digitalocean Key, it takes at least 2 minutes to
create a droplet or snapshot
- String key = System.getenv("DIGITALOCEAN_AUTH_TOKEN");
- if (key != null) {
+ Optional<String> key =
ConfigProvider.getConfig().getOptionalValue("DIGITALOCEAN_AUTH_TOKEN",
String.class);
+ if (key.isPresent()) {
timeoutUnit = TimeUnit.MINUTES;
waitBlockStorageAction = true;
}
diff --git
a/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanTest.java
b/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanTest.java
index 0578458..9bedf96 100644
---
a/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanTest.java
+++
b/integration-tests/digitalocean/src/test/java/org/apache/camel/quarkus/component/digitalocean/it/DigitaloceanTest.java
@@ -26,6 +26,7 @@ import io.quarkus.test.junit.QuarkusTest;
import io.restassured.RestAssured;
import io.restassured.http.ContentType;
import org.apache.camel.quarkus.test.wiremock.MockServer;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.hamcrest.CoreMatchers;
import org.junit.jupiter.api.BeforeAll;
import org.junit.jupiter.api.Test;
@@ -46,12 +47,8 @@ class DigitaloceanTest {
@BeforeAll
public static void initPublicKey() {
- String key = System.getenv("DIGITALOCEAN_PUBLIC_KEY");
- if (key != null) {
- publicKey = key;
- } else {
- publicKey = "ssh-rsa
AEXAMPLEaC1yc2EAAAADAQABAAAAQQDDHr/jh2Jy4yALcK4JyWbVkPRaWmhck3IgCoeOO3z1e2dBowLh64QAM+Qb72pxekALga2oi4GvT+TlWNhzPH4V
example";
- }
+ publicKey =
ConfigProvider.getConfig().getOptionalValue("DIGITALOCEAN_PUBLIC_KEY",
String.class).orElse(
+ "ssh-rsa
AEXAMPLEaC1yc2EAAAADAQABAAAAQQDDHr/jh2Jy4yALcK4JyWbVkPRaWmhck3IgCoeOO3z1e2dBowLh64QAM+Qb72pxekALga2oi4GvT+TlWNhzPH4V
example");
}
@Test
diff --git
a/integration-tests/geocoder/src/main/java/org/apache/camel/quarkus/component/geocoder/it/GeocoderProducers.java
b/integration-tests/geocoder/src/main/java/org/apache/camel/quarkus/component/geocoder/it/GeocoderProducers.java
index b6f24a3..9b4a871 100644
---
a/integration-tests/geocoder/src/main/java/org/apache/camel/quarkus/component/geocoder/it/GeocoderProducers.java
+++
b/integration-tests/geocoder/src/main/java/org/apache/camel/quarkus/component/geocoder/it/GeocoderProducers.java
@@ -16,6 +16,8 @@
*/
package org.apache.camel.quarkus.component.geocoder.it;
+import java.util.Optional;
+
import javax.enterprise.context.ApplicationScoped;
import javax.enterprise.inject.Produces;
import javax.inject.Named;
@@ -23,6 +25,7 @@ import javax.inject.Named;
import io.quarkus.arc.Unremovable;
import org.apache.camel.CamelContext;
import org.apache.camel.component.geocoder.GeoCoderComponent;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.eclipse.microprofile.config.inject.ConfigProperty;
@ApplicationScoped
@@ -43,12 +46,12 @@ public class GeocoderProducers {
@Named("geocoder")
GeoCoderComponent geocoderComponent(CamelContext camelContext,
MockApiService mockApiService)
throws IllegalAccessException, NoSuchFieldException,
InstantiationException {
- final String wireMockUrl = System.getProperty("wiremock.url");
+ final Optional<String> wireMockUrl =
ConfigProvider.getConfig().getOptionalValue("wiremock.url", String.class);
final GeoCoderComponent result = new GeoCoderComponent();
result.setCamelContext(camelContext);
- if (wireMockUrl != null) {
-
result.setGeoApiContext(mockApiService.createGeoApiContext(wireMockUrl,
googleApiKey));
+ if (wireMockUrl.isPresent()) {
+
result.setGeoApiContext(mockApiService.createGeoApiContext(wireMockUrl.get(),
googleApiKey));
}
return result;
}
diff --git
a/integration-tests/lumberjack/src/test/java/org/apache/camel/quarkus/component/lumberjack/it/LumberjackTest.java
b/integration-tests/lumberjack/src/test/java/org/apache/camel/quarkus/component/lumberjack/it/LumberjackTest.java
index 863fb21..a27c298 100644
---
a/integration-tests/lumberjack/src/test/java/org/apache/camel/quarkus/component/lumberjack/it/LumberjackTest.java
+++
b/integration-tests/lumberjack/src/test/java/org/apache/camel/quarkus/component/lumberjack/it/LumberjackTest.java
@@ -22,6 +22,7 @@ import io.quarkus.test.common.QuarkusTestResource;
import io.quarkus.test.common.http.TestHTTPEndpoint;
import io.quarkus.test.junit.QuarkusTest;
import io.restassured.RestAssured;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.junit.jupiter.api.Test;
import static org.hamcrest.Matchers.equalTo;
@@ -94,7 +95,7 @@ class LumberjackTest {
}
private List<LumberjackAckResponse> sendPayload(String portName, boolean
withSsl) throws InterruptedException {
- final int port = Integer.getInteger(portName);
+ final int port = ConfigProvider.getConfig().getValue(portName,
Integer.class);
List<LumberjackAckResponse> ackResponseList =
LumberjackClientUtil.sendMessages(port, withSsl);
return ackResponseList;
}
diff --git
a/integration-tests/slack/src/main/java/org/apache/camel/quarkus/component/slack/it/SlackResource.java
b/integration-tests/slack/src/main/java/org/apache/camel/quarkus/component/slack/it/SlackResource.java
index 2d1ed53..1badb99 100644
---
a/integration-tests/slack/src/main/java/org/apache/camel/quarkus/component/slack/it/SlackResource.java
+++
b/integration-tests/slack/src/main/java/org/apache/camel/quarkus/component/slack/it/SlackResource.java
@@ -31,24 +31,28 @@ import javax.ws.rs.core.Response;
import com.slack.api.model.Message;
import org.apache.camel.ConsumerTemplate;
import org.apache.camel.ProducerTemplate;
+import org.eclipse.microprofile.config.inject.ConfigProperty;
@Path("/slack")
@ApplicationScoped
public class SlackResource {
-
- private static final String SLACK_AUTH_PARAMS =
"serverUrl={{sys:slack.server-url}}&token={{sys:slack.token}}";
-
@Inject
ProducerTemplate producerTemplate;
@Inject
ConsumerTemplate consumerTemplate;
+ @ConfigProperty(name = "slack.server-url")
+ String slackServerUrl;
+
+ @ConfigProperty(name = "slack.token")
+ String slackToken;
+
@Path("/messages")
@GET
@Produces(MediaType.TEXT_PLAIN)
public String getSlackMessages() throws Exception {
- Message message =
consumerTemplate.receiveBody("slack://general?maxResults=1&" +
SLACK_AUTH_PARAMS,
+ Message message =
consumerTemplate.receiveBody("slack://general?maxResults=1&" +
getSlackAuthParams(),
5000L, Message.class);
return message.getText();
}
@@ -57,9 +61,13 @@ public class SlackResource {
@POST
@Consumes(MediaType.TEXT_PLAIN)
public Response createSlackMessage(String message) throws Exception {
- producerTemplate.requestBody("slack://general?" + SLACK_AUTH_PARAMS,
message);
+ producerTemplate.requestBody("slack://general?" +
getSlackAuthParams(), message);
return Response
.created(new URI("https://camel.apache.org/";))
.build();
}
+
+ private String getSlackAuthParams() {
+ return String.format("serverUrl=%s&token=%s", slackServerUrl,
slackToken);
+ }
}
diff --git
a/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTest.java
b/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTest.java
index c7b23b9..c3e8bab 100644
---
a/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTest.java
+++
b/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTest.java
@@ -18,10 +18,13 @@ package org.apache.camel.quarkus.component.slack.it;
import java.util.UUID;
+import com.github.tomakehurst.wiremock.WireMockServer;
import io.quarkus.test.common.QuarkusTestResource;
import io.quarkus.test.junit.QuarkusTest;
import io.restassured.RestAssured;
import io.restassured.http.ContentType;
+import org.apache.camel.quarkus.test.wiremock.MockServer;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.junit.jupiter.api.Test;
import static org.hamcrest.CoreMatchers.equalTo;
@@ -39,6 +42,9 @@ import static org.hamcrest.CoreMatchers.equalTo;
@QuarkusTestResource(SlackTestResource.class)
class SlackTest {
+ @MockServer
+ WireMockServer server;
+
@Test
public void testSlackProduceConsumeMessages() {
final String message = "Hello Camel Quarkus Slack" +
(externalSlackEnabled() ? " " + UUID.randomUUID() : "");
@@ -56,6 +62,6 @@ class SlackTest {
}
boolean externalSlackEnabled() {
- return System.getProperty("wiremock.url") == null;
+ return !ConfigProvider.getConfig().getOptionalValue("wiremock.url",
String.class).isPresent();
}
}
diff --git
a/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTestResource.java
b/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTestResource.java
index 711362f..ee51939 100644
---
a/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTestResource.java
+++
b/integration-tests/slack/src/test/java/org/apache/camel/quarkus/component/slack/it/SlackTestResource.java
@@ -20,6 +20,7 @@ import java.util.Map;
import
org.apache.camel.quarkus.test.wiremock.WireMockTestResourceLifecycleManager;
import org.apache.camel.util.CollectionHelper;
+import org.eclipse.microprofile.config.ConfigProvider;
public class SlackTestResource extends WireMockTestResourceLifecycleManager {
@@ -32,8 +33,10 @@ public class SlackTestResource extends
WireMockTestResourceLifecycleManager {
public Map<String, String> start() {
Map<String, String> properties = super.start();
String wiremockUrl = properties.get("wiremock.url");
- String serverUrl = wiremockUrl != null ? wiremockUrl :
System.getenv(SLACK_ENV_SERVER_URL);
- String webhookUrl = wiremockUrl != null ? wiremockUrl +
"/services/webhook" : System.getenv(SLACK_ENV_WEBHOOK_URL);
+ String serverUrl = wiremockUrl != null ? wiremockUrl
+ : ConfigProvider.getConfig().getValue(SLACK_ENV_SERVER_URL,
String.class);
+ String webhookUrl = wiremockUrl != null ? wiremockUrl +
"/services/webhook"
+ : ConfigProvider.getConfig().getValue(SLACK_ENV_WEBHOOK_URL,
String.class);
return CollectionHelper.mergeMaps(properties, CollectionHelper.mapOf(
"camel.component.slack.webhook-url", webhookUrl,
"slack.server-url", serverUrl,
diff --git
a/integration-tests/splunk/src/test/java/org/apache/camel/quarkus/component/splunk/it/SplunkTest.java
b/integration-tests/splunk/src/test/java/org/apache/camel/quarkus/component/splunk/it/SplunkTest.java
index b1ad98f..cdc350a 100644
---
a/integration-tests/splunk/src/test/java/org/apache/camel/quarkus/component/splunk/it/SplunkTest.java
+++
b/integration-tests/splunk/src/test/java/org/apache/camel/quarkus/component/splunk/it/SplunkTest.java
@@ -27,6 +27,7 @@ import io.quarkus.test.junit.QuarkusTest;
import io.restassured.RestAssured;
import io.restassured.http.ContentType;
import org.apache.camel.util.CollectionHelper;
+import org.eclipse.microprofile.config.ConfigProvider;
import org.junit.Assert;
import org.junit.jupiter.api.Test;
@@ -95,7 +96,7 @@ class SplunkTest {
//create saved search
RestAssured.given()
.baseUri("http://localhost";)
-
.port(Integer.parseInt(System.getProperty(SplunkResource.PARAM_REMOTE_PORT)))
+
.port(ConfigProvider.getConfig().getValue(SplunkResource.PARAM_REMOTE_PORT,
Integer.class))
.contentType(ContentType.JSON)
.param("name", SplunkTestResource.SAVED_SEARCH_NAME)
.param("disabled", "0")
diff --git
a/integration-tests/twilio/src/main/java/org/apache/camel/quarkus/component/twilio/it/TwilioResource.java
b/integration-tests/twilio/src/main/java/org/apache/camel/quarkus/component/twilio/it/TwilioResource.java
index e53e28d..1f2b065 100644
---
a/integration-tests/twilio/src/main/java/org/apache/camel/quarkus/component/twilio/it/TwilioResource.java
+++
b/integration-tests/twilio/src/main/java/org/apache/camel/quarkus/component/twilio/it/TwilioResource.java
@@ -18,6 +18,7 @@ package org.apache.camel.quarkus.component.twilio.it;
import java.util.List;
import java.util.Map;
+import java.util.Optional;
import javax.inject.Inject;
import javax.inject.Named;
@@ -38,6 +39,7 @@ import com.twilio.rest.api.v2010.account.IncomingPhoneNumber;
import com.twilio.rest.api.v2010.account.Message;
import io.quarkus.arc.Unremovable;
import org.apache.camel.ProducerTemplate;
+import org.eclipse.microprofile.config.ConfigProvider;
@Path("/twilio")
public class TwilioResource {
@@ -80,15 +82,15 @@ public class TwilioResource {
@Named("restClient")
public TwilioRestClient restClient() {
// If mocking is enabled, we need to ensure Twilio API calls are
directed to the mock server
- String wireMockUrl = System.getProperty("wiremock.url");
- if (wireMockUrl != null) {
+ Optional<String> wireMockUrl =
ConfigProvider.getConfig().getOptionalValue("wiremock.url", String.class);
+ if (wireMockUrl.isPresent()) {
HttpClient client = new NetworkHttpClient() {
@Override
public com.twilio.http.Response makeRequest(Request
originalRequest) {
String url = originalRequest.getUrl();
Request modified = new Request(originalRequest.getMethod(),
- url.replace("https://api.twilio.com";,
wireMockUrl));
+ url.replace("https://api.twilio.com";,
wireMockUrl.get()));
Map<String, List<String>> headerParams =
originalRequest.getHeaderParams();
for (String key : headerParams.keySet()) {
@@ -118,9 +120,9 @@ public class TwilioResource {
};
return new TwilioRestClient.Builder(
- System.getProperty("camel.component.twilio.username"),
- System.getProperty("camel.component.twilio.password"))
-
.accountSid(System.getProperty("camel.component.twilio.account-sid"))
+
ConfigProvider.getConfig().getValue("camel.component.twilio.username",
String.class),
+
ConfigProvider.getConfig().getValue("camel.component.twilio.password",
String.class))
+
.accountSid(ConfigProvider.getConfig().getValue("camel.component.twilio.account-sid",
String.class))
.httpClient(client)
.build();
}
diff --git
a/integration-tests/vertx-kafka/src/main/java/org/apache/camel/quarkus/component/vertx/kafka/it/VertxKafkaProducers.java
b/integration-tests/vertx-kafka/src/main/java/org/apache/camel/quarkus/component/vertx/kafka/it/VertxKafkaProducers.java
index 6674b85..0b88d87 100644
---
a/integration-tests/vertx-kafka/src/main/java/org/apache/camel/quarkus/component/vertx/kafka/it/VertxKafkaProducers.java
+++
b/integration-tests/vertx-kafka/src/main/java/org/apache/camel/quarkus/component/vertx/kafka/it/VertxKafkaProducers.java
@@ -30,6 +30,7 @@ import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.internals.DefaultPartitioner;
import org.apache.kafka.common.serialization.StringDeserializer;
import org.apache.kafka.common.serialization.StringSerializer;
+import org.eclipse.microprofile.config.ConfigProvider;
public class VertxKafkaProducers {
@@ -44,7 +45,7 @@ public class VertxKafkaProducers {
config.put(ConsumerConfig.AUTO_COMMIT_INTERVAL_MS_CONFIG, "1000");
config.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest");
config.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
-
System.getProperty("camel.component.vertx-kafka.bootstrap-servers"));
+
ConfigProvider.getConfig().getValue("camel.component.vertx-kafka.bootstrap-servers",
String.class));
config.put(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, "true");
config.put(ConsumerConfig.GROUP_ID_CONFIG, GROUP_ID);
config.put(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG,
StringDeserializer.class.getName());
@@ -62,7 +63,7 @@ public class VertxKafkaProducers {
Map<String, String> config = new HashMap<>();
config.put(ProducerConfig.ACKS_CONFIG, "1");
config.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
-
System.getProperty("camel.component.vertx-kafka.bootstrap-servers"));
+
ConfigProvider.getConfig().getValue("camel.component.vertx-kafka.bootstrap-servers",
String.class));
config.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
config.put(ProducerConfig.PARTITIONER_CLASS_CONFIG,
DefaultPartitioner.class.getName());
config.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
StringSerializer.class.getName());
