-
Notifications
You must be signed in to change notification settings - Fork 217
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Allow to configure the consumer group ID for a kafka connection
Signed-off-by: Yannic Klem <yannic.klem@bosch.io>
- Loading branch information
Showing
3 changed files
with
212 additions
and
5 deletions.
There are no files selected for viewing
85 changes: 85 additions & 0 deletions
85
.../eclipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerGroupSpecificConfig.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,85 @@ | ||
| /* | ||
| * 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.messaging.kafka; | ||
|
|
||
| import java.text.MessageFormat; | ||
| import java.util.Map; | ||
| import java.util.Optional; | ||
| import java.util.regex.Pattern; | ||
|
|
||
| import org.apache.kafka.clients.consumer.ConsumerConfig; | ||
| import org.eclipse.ditto.base.model.headers.DittoHeaders; | ||
| import org.eclipse.ditto.connectivity.model.Connection; | ||
| import org.eclipse.ditto.connectivity.model.ConnectionConfigurationInvalidException; | ||
|
|
||
| /** | ||
| * Allows to configure a consumer group ID via the specific config of a connection. | ||
| */ | ||
| final class KafkaConsumerGroupSpecificConfig implements KafkaSpecificConfig { | ||
|
|
||
| private static final String GROUP_ID_ALLOWED_CHARACTERS = "[a-zA-Z0-9\\._\\-]"; | ||
| private static final Pattern GROUP_ID_VALIDATION_PATTERN = Pattern.compile(GROUP_ID_ALLOWED_CHARACTERS + "+"); | ||
| private static final String GROUP_ID_SPECIFIC_CONFIG_KEY = "groupId"; | ||
|
|
||
| private static KafkaConsumerGroupSpecificConfig instance; | ||
|
|
||
| private KafkaConsumerGroupSpecificConfig() { | ||
| } | ||
|
|
||
| static KafkaConsumerGroupSpecificConfig getInstance() { | ||
| if (instance == null) { | ||
| instance = new KafkaConsumerGroupSpecificConfig(); | ||
| } | ||
| return instance; | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isApplicable(final Connection connection) { | ||
| return !connection.getSources().isEmpty(); | ||
| } | ||
|
|
||
| @Override | ||
| public void validateOrThrow(final Connection connection, final DittoHeaders dittoHeaders) { | ||
| if (!isValid(connection)) { | ||
| final String groupId = getGroupId(connection).orElse(""); | ||
| final String message = MessageFormat.format( | ||
| "The connection configuration contains an invalid value for the consumer group ID. " + | ||
| "Allowed Characters are: <{1}>", groupId, GROUP_ID_ALLOWED_CHARACTERS); | ||
| throw ConnectionConfigurationInvalidException.newBuilder(message) | ||
| .dittoHeaders(dittoHeaders) | ||
| .build(); | ||
| } | ||
| } | ||
|
|
||
| @Override | ||
| public boolean isValid(final Connection connection) { | ||
| final Optional<String> optionalGroupId = getGroupId(connection); | ||
| if (optionalGroupId.isPresent()) { | ||
| final String groupId = optionalGroupId.get(); | ||
| return GROUP_ID_VALIDATION_PATTERN.matcher(groupId).matches(); | ||
| } | ||
| return true; | ||
| } | ||
|
|
||
| @Override | ||
| public Map<String, String> apply(final Connection connection) { | ||
| return getGroupId(connection) | ||
| .map(groupId -> Map.of(ConsumerConfig.GROUP_ID_CONFIG, groupId)) | ||
| .orElseGet(Map::of); | ||
| } | ||
|
|
||
| private Optional<String> getGroupId(final Connection connection) { | ||
| return Optional.ofNullable(connection.getSpecificConfig().get(GROUP_ID_SPECIFIC_CONFIG_KEY)); | ||
| } | ||
|
|
||
| } |
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
103 changes: 103 additions & 0 deletions
103
...ipse/ditto/connectivity/service/messaging/kafka/KafkaConsumerGroupSpecificConfigTest.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,103 @@ | ||
| /* | ||
| * 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.messaging.kafka; | ||
|
|
||
| import static org.assertj.core.api.Assertions.assertThat; | ||
| import static org.assertj.core.api.Assertions.assertThatCode; | ||
| import static org.mockito.Mockito.mock; | ||
| import static org.mockito.Mockito.when; | ||
|
|
||
| import java.util.List; | ||
| import java.util.Map; | ||
|
|
||
| import org.apache.kafka.clients.consumer.ConsumerConfig; | ||
| import org.eclipse.ditto.base.model.headers.DittoHeaders; | ||
| import org.eclipse.ditto.connectivity.model.Connection; | ||
| import org.eclipse.ditto.connectivity.model.ConnectionConfigurationInvalidException; | ||
| import org.eclipse.ditto.connectivity.model.Source; | ||
| import org.junit.Test; | ||
|
|
||
| public final class KafkaConsumerGroupSpecificConfigTest { | ||
|
|
||
| private final KafkaConsumerGroupSpecificConfig underTest = KafkaConsumerGroupSpecificConfig.getInstance(); | ||
|
|
||
| @Test | ||
| public void isNotApplicableToConnectionWithoutSources() { | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSources()).thenReturn(List.of()); | ||
| assertThat(underTest.isApplicable(connection)).isFalse(); | ||
| } | ||
|
|
||
| @Test | ||
| public void isApplicableToConnectionWithSources() { | ||
| final Connection connection = mock(Connection.class); | ||
| final Source source = mock(Source.class); | ||
| when(connection.getSources()).thenReturn(List.of(source)); | ||
| assertThat(underTest.isApplicable(connection)).isTrue(); | ||
| } | ||
|
|
||
| @Test | ||
| public void invalidCharactersCauseConnectionConfigurationInvalidException() { | ||
| final Map<String, String> specificConfig = Map.of("groupId", "invalidCharacterÜ"); | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSpecificConfig()).thenReturn(specificConfig); | ||
| final DittoHeaders dittoHeaders = DittoHeaders.empty(); | ||
| assertThatCode(() -> underTest.validateOrThrow(connection, dittoHeaders)) | ||
| .isExactlyInstanceOf(ConnectionConfigurationInvalidException.class); | ||
| } | ||
|
|
||
| @Test | ||
| public void validCharactersCauseNoException() { | ||
| final Map<String, String> specificConfig = Map.of("groupId", "only-Valid-Characters-1234"); | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSpecificConfig()).thenReturn(specificConfig); | ||
| final DittoHeaders dittoHeaders = DittoHeaders.empty(); | ||
| assertThatCode(() -> underTest.validateOrThrow(connection, dittoHeaders)) | ||
| .doesNotThrowAnyException(); | ||
| } | ||
|
|
||
| @Test | ||
| public void invalidCharactersAreInvalid() { | ||
| final Map<String, String> specificConfig = Map.of("groupId", "invalidCharacterÜ"); | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSpecificConfig()).thenReturn(specificConfig); | ||
| assertThat(underTest.isValid(connection)).isFalse(); | ||
| } | ||
|
|
||
| @Test | ||
| public void validCharactersAreValid() { | ||
| final Map<String, String> specificConfig = Map.of("groupId", "only-Valid-Characters-1234"); | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSpecificConfig()).thenReturn(specificConfig); | ||
| assertThat(underTest.isValid(connection)).isTrue(); | ||
| } | ||
|
|
||
| @Test | ||
| public void emptyGroupIdIsInvalid() { | ||
| final Map<String, String> specificConfig = Map.of("groupId", ""); | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSpecificConfig()).thenReturn(specificConfig); | ||
| assertThat(underTest.isValid(connection)).isFalse(); | ||
| } | ||
|
|
||
| @Test | ||
| public void applyReturnsGroupIdForConsumerConfigKey() { | ||
| final Map<String, String> specificConfig = Map.of("groupId", "only-Valid-Characters-1234"); | ||
| final Connection connection = mock(Connection.class); | ||
| when(connection.getSpecificConfig()).thenReturn(specificConfig); | ||
|
|
||
| final Map<String, String> expectedConfig = Map.of(ConsumerConfig.GROUP_ID_CONFIG, "only-Valid-Characters-1234"); | ||
| assertThat(underTest.apply(connection)).isEqualTo(expectedConfig); | ||
| } | ||
|
|
||
| } |