You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@streampipes.apache.org by ze...@apache.org on 2021/08/24 06:33:50 UTC
[incubator-streampipes] branch dev updated: [STREAMPIPES-416] Add
environment variable Kafka retention time of producers
This is an automated email from the ASF dual-hosted git repository.
zehnder pushed a commit to branch dev
in repository https://gitbox.apache.org/repos/asf/incubator-streampipes.git
The following commit(s) were added to refs/heads/dev by this push:
new 2bea25f [STREAMPIPES-416] Add environment variable Kafka retention time of producers
new 362fe0d Merge branch 'dev' of github.com:apache/incubator-streampipes into dev
2bea25f is described below
commit 2bea25f3675adc2784a8984cf524b9636584dc0c
Author: Philipp Zehnder <ze...@fzi.de>
AuthorDate: Tue Aug 24 08:33:03 2021 +0200
[STREAMPIPES-416] Add environment variable Kafka retention time of producers
---
.../main/java/org/apache/streampipes/commons/constants/Envs.java | 3 ++-
streampipes-messaging-kafka/pom.xml | 5 +++++
.../org/apache/streampipes/messaging/kafka/SpKafkaProducer.java | 6 +++++-
3 files changed, 12 insertions(+), 2 deletions(-)
diff --git a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
index dac07be..c37c12b 100644
--- a/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
+++ b/streampipes-commons/src/main/java/org/apache/streampipes/commons/constants/Envs.java
@@ -21,7 +21,8 @@ public enum Envs {
SP_HOST("SP_HOST"),
SP_PORT("SP_PORT"),
- SP_CONSUL_LOCATION("CONSUL_LOCATION");
+ SP_CONSUL_LOCATION("CONSUL_LOCATION"),
+ SP_KAFKA_RETENTION_MS("SP_KAFKA_RETENTION_MS");
private final String envVariableName;
diff --git a/streampipes-messaging-kafka/pom.xml b/streampipes-messaging-kafka/pom.xml
index e828da1..e48055d 100644
--- a/streampipes-messaging-kafka/pom.xml
+++ b/streampipes-messaging-kafka/pom.xml
@@ -34,6 +34,11 @@
<artifactId>streampipes-messaging</artifactId>
<version>0.69.0-SNAPSHOT</version>
</dependency>
+ <dependency>
+ <groupId>org.apache.streampipes</groupId>
+ <artifactId>streampipes-commons</artifactId>
+ <version>0.69.0-SNAPSHOT</version>
+ </dependency>
<!-- External dependencies -->
<dependency>
diff --git a/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java b/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java
index 6990644..bfca8a0 100644
--- a/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java
+++ b/streampipes-messaging-kafka/src/main/java/org/apache/streampipes/messaging/kafka/SpKafkaProducer.java
@@ -27,6 +27,7 @@ import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.config.TopicConfig;
+import org.apache.streampipes.commons.constants.Envs;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.apache.streampipes.messaging.EventProducer;
@@ -42,7 +43,9 @@ import java.util.concurrent.ExecutionException;
public class SpKafkaProducer implements EventProducer<KafkaTransportProtocol>, Serializable {
+
private static final String COLON = ":";
+ private static final String SP_KAFKA_RETENTION_MS_DEFAULT = "600000";
private String brokerUrl;
private String topic;
@@ -115,7 +118,8 @@ public class SpKafkaProducer implements EventProducer<KafkaTransportProtocol>, S
props.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, brokerUrl);
Map<String, String> topicConfig = new HashMap<>();
- topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, "600000");
+ String retentionTime = Envs.SP_KAFKA_RETENTION_MS.exists() ? Envs.SP_KAFKA_RETENTION_MS.getValue() : SP_KAFKA_RETENTION_MS_DEFAULT;
+ topicConfig.put(TopicConfig.RETENTION_MS_CONFIG, retentionTime);
AdminClient adminClient = KafkaAdminClient.create(props);