-
Notifications
You must be signed in to change notification settings - Fork 21
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Metrics are not supported in this PR and will be added in another one. Related to #421
- Loading branch information
Showing
20 changed files
with
933 additions
and
4 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
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
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
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
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
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,89 @@ | ||
| ## | ||
| # Copyright 2023 Aiven Oy | ||
| # | ||
| # Licensed under the Apache License, Version 2.0 (the "License"); | ||
| # you may not use this file except in compliance with the License. | ||
| # You may obtain a copy of the License at | ||
| # | ||
| # http://www.apache.org/licenses/LICENSE-2.0 | ||
| # | ||
| # Unless required by applicable law or agreed to in writing, software | ||
| # distributed under the License is distributed on an "AS IS" BASIS, | ||
| # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| # See the License for the specific language governing permissions and | ||
| # limitations under the License. | ||
| ## | ||
| version: '3.8' | ||
| services: | ||
| zookeeper: | ||
| image: "confluentinc/cp-zookeeper:7.3.3" | ||
| ports: | ||
| - "2181:2181" | ||
| environment: | ||
| ZOOKEEPER_CLIENT_PORT: 2181 | ||
|
|
||
| kafka: | ||
| image: "aivenoy/kafka-with-ts-plugin" | ||
| container_name: "kafka-ts" | ||
| depends_on: | ||
| - zookeeper | ||
| - azurite | ||
| ports: | ||
| - "9092:9092" | ||
| - "7000:7000" #prometheus metrics | ||
| environment: | ||
| KAFKA_BROKER_ID: 0 | ||
| KAFKA_ZOOKEEPER_CONNECT: "zookeeper:2181" | ||
| KAFKA_LISTENERS: "PLAINTEXT://0.0.0.0:9092,BROKER://0.0.0.0:29092" | ||
| KAFKA_ADVERTISED_LISTENERS: "PLAINTEXT://localhost:9092,BROKER://kafka:29092" | ||
| KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: "PLAINTEXT:PLAINTEXT,BROKER:PLAINTEXT" | ||
| KAFKA_INTER_BROKER_LISTENER_NAME: "BROKER" | ||
| KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1 | ||
| KAFKA_OFFSETS_TOPIC_NUM_PARTITIONS: 1 | ||
| KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 1 | ||
| KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 1 | ||
| KAFKA_AUTO_CREATE_TOPICS_ENABLE: false | ||
| # Increase Tiered Storage log level | ||
| KAFKA_LOG4J_LOGGERS: "io.aiven.kafka.tieredstorage=DEBUG" | ||
| # Tweak retention checking | ||
| KAFKA_LOG_RETENTION_CHECK_INTERVAL_MS: 10000 | ||
| # Enable Tiered Storage | ||
| KAFKA_REMOTE_LOG_STORAGE_SYSTEM_ENABLE: true | ||
| KAFKA_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS: 5000 | ||
| # Remote metadata manager | ||
| KAFKA_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME: "org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager" | ||
| KAFKA_REMOTE_LOG_METADATA_MANAGER_IMPL_PREFIX: "rlmm.config." | ||
| KAFKA_REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME: "BROKER" | ||
| KAFKA_RLMM_CONFIG_REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR: 1 | ||
| # Remote storage manager | ||
| KAFKA_REMOTE_LOG_STORAGE_MANAGER_CLASS_PATH: "/tiered-storage-for-apache-kafka/core/*:/tiered-storage-for-apache-kafka/azure/*" | ||
| KAFKA_REMOTE_LOG_STORAGE_MANAGER_CLASS_NAME: "io.aiven.kafka.tieredstorage.RemoteStorageManager" | ||
| KAFKA_REMOTE_LOG_STORAGE_MANAGER_IMPL_PREFIX: "rsm.config." | ||
| KAFKA_RSM_CONFIG_CHUNK_SIZE: 5242880 # 5MiB | ||
| KAFKA_RSM_CONFIG_CHUNK_CACHE_CLASS: "io.aiven.kafka.tieredstorage.chunkmanager.cache.InMemoryChunkCache" | ||
| KAFKA_RSM_CONFIG_CHUNK_CACHE_SIZE: -1 | ||
| KAFKA_RSM_CONFIG_CUSTOM_METADATA_FIELDS_INCLUDE: "REMOTE_SIZE" | ||
| # Storage backend | ||
| KAFKA_RSM_CONFIG_KEY_PREFIX: "tiered-storage-demo/" | ||
| KAFKA_RSM_CONFIG_STORAGE_BACKEND_CLASS: "io.aiven.kafka.tieredstorage.storage.azure.AzureBlobStorage" | ||
| KAFKA_RSM_CONFIG_STORAGE_AZURE_CONTAINER_NAME: "test-container" | ||
| KAFKA_RSM_CONFIG_STORAGE_AZURE_ACCOUNT_NAME: "devstoreaccount1" | ||
| KAFKA_RSM_CONFIG_STORAGE_AZURE_ACCOUNT_KEY: "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw==" | ||
| KAFKA_RSM_CONFIG_STORAGE_AZURE_ENDPOINT_URL: "http://azurite:10000/devstoreaccount1" | ||
|
|
||
| azurite: | ||
| image: mcr.microsoft.com/azure-storage/azurite | ||
| ports: | ||
| - "10000:10000" | ||
| command: azurite-blob --blobHost 0.0.0.0 | ||
|
|
||
| azurite-create-container: | ||
| image: mcr.microsoft.com/azure-cli | ||
| restart: "no" | ||
| depends_on: | ||
| - azurite | ||
| command: > | ||
| az storage container create --name test-container | ||
| --account-name devstoreaccount1 | ||
| --account-key Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw== | ||
| --blob-endpoint http://azurite:10000/devstoreaccount1 |
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
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
106 changes: 106 additions & 0 deletions
106
e2e/src/test/java/io/aiven/kafka/tieredstorage/e2e/AzureSingleBrokerTest.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,106 @@ | ||
| /* | ||
| * Copyright 2023 Aiven Oy | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.aiven.kafka.tieredstorage.e2e; | ||
|
|
||
| import java.util.List; | ||
| import java.util.stream.Collectors; | ||
|
|
||
| import org.apache.kafka.common.TopicIdPartition; | ||
| import org.apache.kafka.common.Uuid; | ||
|
|
||
| import com.azure.storage.blob.BlobContainerClient; | ||
| import com.azure.storage.blob.BlobServiceClientBuilder; | ||
| import com.azure.storage.blob.models.BlobItem; | ||
| import com.azure.storage.blob.models.ListBlobsOptions; | ||
| import org.junit.jupiter.api.AfterAll; | ||
| import org.junit.jupiter.api.BeforeAll; | ||
| import org.testcontainers.containers.GenericContainer; | ||
| import org.testcontainers.utility.DockerImageName; | ||
|
|
||
| public class AzureSingleBrokerTest extends SingleBrokerTest { | ||
| private static final int BLOB_STORAGE_PORT = 10000; | ||
| static final String NETWORK_ALIAS = "blob-storage"; | ||
|
|
||
| static final GenericContainer<?> AZURITE_SERVER = | ||
| new GenericContainer<>(DockerImageName.parse("mcr.microsoft.com/azure-storage/azurite")) | ||
| .withExposedPorts(BLOB_STORAGE_PORT) | ||
| .withCommand("azurite-blob --blobHost 0.0.0.0") | ||
| .withNetwork(NETWORK) | ||
| .withNetworkAliases(NETWORK_ALIAS); | ||
|
|
||
| static final String AZURE_CONTAINER = "test-container"; | ||
|
|
||
| static BlobContainerClient blobContainerClient; | ||
|
|
||
| @BeforeAll | ||
| static void init() throws Exception { | ||
| AZURITE_SERVER.start(); | ||
|
|
||
| // The well-known Azurite account name and key. | ||
| final String accountName = "devstoreaccount1"; | ||
| final String accountKey = | ||
| "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; | ||
|
|
||
| final String endpointForTestCode = | ||
| "http://127.0.0.1:" + AZURITE_SERVER.getMappedPort(BLOB_STORAGE_PORT) + "/devstoreaccount1"; | ||
| final String connectionString = "DefaultEndpointsProtocol=http;" | ||
| + "AccountName=" + accountName + ";" | ||
| + "AccountKey=" + accountKey + ";" | ||
| + "BlobEndpoint=" + endpointForTestCode + ";"; | ||
| final var blobServiceClient = new BlobServiceClientBuilder() | ||
| .connectionString(connectionString) | ||
| .buildClient(); | ||
| blobContainerClient = blobServiceClient.createBlobContainer(AZURE_CONTAINER); | ||
|
|
||
| final String endpointForKafka = "http://" + NETWORK_ALIAS + ":" + BLOB_STORAGE_PORT + "/devstoreaccount1"; | ||
| setupKafka(kafka -> kafka.withEnv("KAFKA_RSM_CONFIG_STORAGE_BACKEND_CLASS", | ||
| "io.aiven.kafka.tieredstorage.storage.azure.AzureBlobStorage") | ||
| .withEnv("KAFKA_REMOTE_LOG_STORAGE_MANAGER_CLASS_PATH", | ||
| "/tiered-storage-for-apache-kafka/core/*:/tiered-storage-for-apache-kafka/azure/*") | ||
| .withEnv("KAFKA_RSM_CONFIG_STORAGE_AZURE_CONTAINER_NAME", AZURE_CONTAINER) | ||
| .withEnv("KAFKA_RSM_CONFIG_STORAGE_AZURE_ACCOUNT_NAME", accountName) | ||
| .withEnv("KAFKA_RSM_CONFIG_STORAGE_AZURE_ACCOUNT_KEY", accountKey) | ||
| .withEnv("KAFKA_RSM_CONFIG_STORAGE_AZURE_ENDPOINT_URL", endpointForKafka) | ||
| .dependsOn(AZURITE_SERVER)); | ||
| } | ||
|
|
||
| @AfterAll | ||
| static void cleanup() { | ||
| stopKafka(); | ||
|
|
||
| AZURITE_SERVER.stop(); | ||
|
|
||
| cleanupStorage(); | ||
| } | ||
|
|
||
| @Override | ||
| boolean assertNoTopicDataOnTierStorage(final String topicName, final Uuid topicId) { | ||
| final String prefix = String.format("%s-%s", topicName, topicId.toString()); | ||
|
|
||
| final var list = blobContainerClient.listBlobs(new ListBlobsOptions().setPrefix(prefix), null); | ||
| return list.stream().findAny().isEmpty(); | ||
| } | ||
|
|
||
| @Override | ||
| List<String> remotePartitionFiles(final TopicIdPartition topicIdPartition) { | ||
| return blobContainerClient.listBlobs().stream() | ||
| .map(BlobItem::getName) | ||
| .map(k -> k.substring(k.lastIndexOf('/') + 1)) | ||
| .sorted() | ||
| .collect(Collectors.toList()); | ||
| } | ||
| } |
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
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,35 @@ | ||
| /* | ||
| * Copyright 2023 Aiven Oy | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| archivesBaseName = "storage-azure" | ||
|
|
||
| dependencies { | ||
| implementation project(":storage:core") | ||
|
|
||
| implementation platform("com.azure:azure-sdk-bom:$azureSdkVersion") | ||
| implementation ("com.azure:azure-identity") { | ||
| exclude group: "org.slf4j" | ||
| } | ||
| implementation ("com.azure:azure-storage-blob") { | ||
| exclude group: "org.slf4j" | ||
| } | ||
|
|
||
| implementation project(":commons") | ||
|
|
||
| testImplementation(testFixtures(project(":storage:core"))) | ||
|
|
||
| testImplementation "org.testcontainers:junit-jupiter:$testcontainersVersion" | ||
| } |
40 changes: 40 additions & 0 deletions
40
...-test/java/io/aiven/kafka/tieredstorage/storage/azure/AzureBlobStorageAccountKeyTest.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,40 @@ | ||
| /* | ||
| * Copyright 2023 Aiven Oy | ||
| * | ||
| * Licensed under the Apache License, Version 2.0 (the "License"); | ||
| * you may not use this file except in compliance with the License. | ||
| * You may obtain a copy of the License at | ||
| * | ||
| * http://www.apache.org/licenses/LICENSE-2.0 | ||
| * | ||
| * Unless required by applicable law or agreed to in writing, software | ||
| * distributed under the License is distributed on an "AS IS" BASIS, | ||
| * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. | ||
| * See the License for the specific language governing permissions and | ||
| * limitations under the License. | ||
| */ | ||
|
|
||
| package io.aiven.kafka.tieredstorage.storage.azure; | ||
|
|
||
| import java.util.Map; | ||
|
|
||
| import io.aiven.kafka.tieredstorage.storage.StorageBackend; | ||
|
|
||
| class AzureBlobStorageAccountKeyTest extends AzureBlobStorageTest { | ||
| @Override | ||
| protected StorageBackend storage() { | ||
| final AzureBlobStorage azureBlobStorage = new AzureBlobStorage(); | ||
| // The well-known Azurite account name and key. | ||
| final String accountName = "devstoreaccount1"; | ||
| final String accountKey = | ||
| "Eby8vdM02xNOcqFlqUwJPLlmEtlCDXJ1OUzFT50uSRZ6IFsuFq2UVErCz4I6tq/K1SZFPTOtr/KBHBeksoGMGw=="; | ||
| final Map<String, Object> configs = Map.of( | ||
| "azure.container.name", azureContainerName, | ||
| "azure.account.name", accountName, | ||
| "azure.account.key", accountKey, | ||
| "azure.endpoint.url", endpoint() | ||
| ); | ||
| azureBlobStorage.configure(configs); | ||
| return azureBlobStorage; | ||
| } | ||
| } |
Oops, something went wrong.