Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
#586 make kafka producer restart backoff configurable, move config fo…
…r alpakka into separate config inside of consumer and producer, implemented config model for consumer and producer Signed-off-by: Johannes Schneider <johannes.schneider@bosch.io>
- Loading branch information
Showing
20 changed files
with
655 additions
and
317 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
84 changes: 84 additions & 0 deletions
84
...c/main/java/org/eclipse/ditto/connectivity/service/config/DefaultKafkaConsumerConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,84 @@ | ||
| /* | ||
| * Copyright (c) 2021 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.connectivity.service.config; | ||
|
|
||
| import java.util.Objects; | ||
|
|
||
| import javax.annotation.concurrent.Immutable; | ||
|
|
||
| import org.eclipse.ditto.base.service.config.ThrottlingConfig; | ||
|
|
||
| import com.typesafe.config.Config; | ||
| import com.typesafe.config.ConfigFactory; | ||
|
|
||
| /** | ||
| * This class is the default implementation of {@link KafkaConsumerConfig}. | ||
| */ | ||
| @Immutable | ||
| public final class DefaultKafkaConsumerConfig implements KafkaConsumerConfig { | ||
|
|
||
| private static final String ALPAKKA_PATH = "alpakka"; | ||
|
|
||
| private final ThrottlingConfig throttlingConfig; | ||
| private final Config alpakkaConfig; | ||
|
|
||
| private DefaultKafkaConsumerConfig(final Config kafkaConsumerScopedConfig) { | ||
| throttlingConfig = ThrottlingConfig.of(kafkaConsumerScopedConfig.hasPath(ThrottlingConfig.CONFIG_PATH) | ||
| ? kafkaConsumerScopedConfig.getConfig(ThrottlingConfig.CONFIG_PATH) | ||
| : ConfigFactory.empty()); | ||
| alpakkaConfig = kafkaConsumerScopedConfig.getConfig(ALPAKKA_PATH); | ||
| } | ||
|
|
||
| /** | ||
| * Returns an instance of {@code DefaultKafkaConsumerConfig} based on the settings of the specified Config. | ||
| * | ||
| * @param config is supposed to provide the Kafka config setting. | ||
| * @return the instance. | ||
| * @throws org.eclipse.ditto.internal.utils.config.DittoConfigError if {@code config} is invalid. | ||
| */ | ||
| public static DefaultKafkaConsumerConfig of(final Config config) { | ||
| return new DefaultKafkaConsumerConfig(config); | ||
| } | ||
|
|
||
| @Override | ||
| public ThrottlingConfig getThrottlingConfig() { | ||
| return throttlingConfig; | ||
| } | ||
|
|
||
| @Override | ||
| public Config getAlpakkaConfig() { | ||
| return alpakkaConfig; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(final Object o) { | ||
| if (this == o) return true; | ||
| if (o == null || getClass() != o.getClass()) return false; | ||
| final DefaultKafkaConsumerConfig that = (DefaultKafkaConsumerConfig) o; | ||
| return Objects.equals(throttlingConfig, that.throttlingConfig) && | ||
| Objects.equals(alpakkaConfig, that.alpakkaConfig); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(throttlingConfig, alpakkaConfig); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return getClass().getSimpleName() + " [" + | ||
| "throttlingConfig=" + throttlingConfig + | ||
| ", alpakkaConfig=" + alpakkaConfig + | ||
| "]"; | ||
| } | ||
| } |
121 changes: 121 additions & 0 deletions
121
...c/main/java/org/eclipse/ditto/connectivity/service/config/DefaultKafkaProducerConfig.java
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,121 @@ | ||
| /* | ||
| * Copyright (c) 2021 Contributors to the Eclipse Foundation | ||
| * | ||
| * See the NOTICE file(s) distributed with this work for additional | ||
| * information regarding copyright ownership. | ||
| * | ||
| * This program and the accompanying materials are made available under the | ||
| * terms of the Eclipse Public License 2.0 which is available at | ||
| * http://www.eclipse.org/legal/epl-2.0 | ||
| * | ||
| * SPDX-License-Identifier: EPL-2.0 | ||
| */ | ||
| package org.eclipse.ditto.connectivity.service.config; | ||
|
|
||
| import java.time.Duration; | ||
| import java.util.Objects; | ||
|
|
||
| import javax.annotation.concurrent.Immutable; | ||
|
|
||
| import com.typesafe.config.Config; | ||
|
|
||
| /** | ||
| * This class is the default implementation of {@link KafkaProducerConfig}. | ||
| */ | ||
| @Immutable | ||
| public final class DefaultKafkaProducerConfig implements KafkaProducerConfig { | ||
|
|
||
| private static final String QUEUE_SIZE_PATH = "queue-size"; | ||
| private static final String PARALLELISM_PATH = "parallelism"; | ||
| private static final String MIN_BACKOFF_PATH = "min-backoff"; | ||
| private static final String MAX_BACKOFF_PATH = "max-backoff"; | ||
| private static final String RANDOM_FACTOR_PATH = "random-factor"; | ||
| private static final String ALPAKKA_PATH = "alpakka"; | ||
|
|
||
| private final int queueSize; | ||
| private final int parallelism; | ||
| private final Duration minBackoff; | ||
| private final Duration maxBackoff; | ||
| private final double randomFactor; | ||
| private final Config alpakkaConfig; | ||
|
|
||
| private DefaultKafkaProducerConfig(final Config kafkaProducerScopedConfig) { | ||
| queueSize = kafkaProducerScopedConfig.getInt(QUEUE_SIZE_PATH); | ||
| parallelism = kafkaProducerScopedConfig.getInt(PARALLELISM_PATH); | ||
| minBackoff = kafkaProducerScopedConfig.getDuration(MIN_BACKOFF_PATH); | ||
| maxBackoff = kafkaProducerScopedConfig.getDuration(MAX_BACKOFF_PATH); | ||
| randomFactor = kafkaProducerScopedConfig.getDouble(RANDOM_FACTOR_PATH); | ||
| alpakkaConfig = kafkaProducerScopedConfig.getConfig(ALPAKKA_PATH); | ||
| } | ||
|
|
||
| /** | ||
| * Returns an instance of {@code DefaultKafkaProducerConfig} based on the settings of the specified Config. | ||
| * | ||
| * @param config is supposed to provide the Kafka config setting. | ||
| * @return the instance. | ||
| * @throws org.eclipse.ditto.internal.utils.config.DittoConfigError if {@code config} is invalid. | ||
| */ | ||
| public static DefaultKafkaProducerConfig of(final Config config) { | ||
| return new DefaultKafkaProducerConfig(config); | ||
| } | ||
|
|
||
| @Override | ||
| public int getQueueSize() { | ||
| return queueSize; | ||
| } | ||
|
|
||
| @Override | ||
| public int getParallelism() { | ||
| return parallelism; | ||
| } | ||
|
|
||
| @Override | ||
| public Duration getMinBackoff() { | ||
| return minBackoff; | ||
| } | ||
|
|
||
| @Override | ||
| public Duration getMaxBackoff() { | ||
| return maxBackoff; | ||
| } | ||
|
|
||
| @Override | ||
| public double getRandomFactor() { | ||
| return randomFactor; | ||
| } | ||
|
|
||
| @Override | ||
| public Config getAlpakkaConfig() { | ||
| return alpakkaConfig; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean equals(final Object o) { | ||
| if (this == o) return true; | ||
| if (o == null || getClass() != o.getClass()) return false; | ||
| final DefaultKafkaProducerConfig that = (DefaultKafkaProducerConfig) o; | ||
| return Objects.equals(queueSize, that.queueSize) && | ||
| Objects.equals(parallelism, that.parallelism) && | ||
| Objects.equals(minBackoff, that.minBackoff) && | ||
| Objects.equals(maxBackoff, that.maxBackoff) && | ||
| Objects.equals(randomFactor, that.randomFactor) && | ||
| Objects.equals(alpakkaConfig, that.alpakkaConfig); | ||
| } | ||
|
|
||
| @Override | ||
| public int hashCode() { | ||
| return Objects.hash(queueSize, parallelism, minBackoff, maxBackoff, randomFactor, alpakkaConfig); | ||
| } | ||
|
|
||
| @Override | ||
| public String toString() { | ||
| return getClass().getSimpleName() + " [" + | ||
| "queueSize=" + queueSize + | ||
| ", parallelism=" + parallelism + | ||
| ", minBackoff=" + minBackoff + | ||
| ", maxBackoff=" + maxBackoff + | ||
| ", randomFactor=" + randomFactor + | ||
| ", alpakkaConfig=" + alpakkaConfig + | ||
| "]"; | ||
| } | ||
| } |
Oops, something went wrong.