Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

KAFKA-16882: Migrate RemoteLogSegmentLifecycleTest to ClusterInstance infra #16180

Merged
merged 3 commits into from
Jun 4, 2024
Merged

KAFKA-16882: Migrate RemoteLogSegmentLifecycleTest to ClusterInstance infra #16180

Show file tree
Hide file tree
Changes from 2 commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
b87fe1d
KAFKA-16882: Migrate RemoteLogSegmentLifecycleTest to ClusterInstance…
kamalcph Jun 3, 2024
b95f6ce
Addressed the review comments
kamalcph Jun 4, 2024
49e8dcd
Addressed the review comments.
kamalcph Jun 4, 2024
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
23 changes: 2 additions & 21 deletions ...rg/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataManagerTestUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,17 +17,12 @@
package org.apache.kafka.server.log.remote.metadata.storage;

import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.test.TestUtils;
import org.junit.jupiter.api.Assertions;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Objects;
import java.util.Set;
import java.util.function.Function;
import java.util.function.Supplier;

Expand All @@ -37,10 +32,9 @@
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP;
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR_PROP;
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.REMOTE_LOG_METADATA_TOPIC_RETENTION_MS_PROP;
import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;

public class RemoteLogMetadataManagerTestUtils {
private static final Logger log = LoggerFactory.getLogger(RemoteLogMetadataManagerTestUtils.class);

private static final int METADATA_TOPIC_PARTITIONS_COUNT = 3;
private static final short METADATA_TOPIC_REPLICATION_FACTOR = 2;
private static final long METADATA_TOPIC_RETENTION_MS = 24 * 60 * 60 * 1000L;
Expand All @@ -53,7 +47,6 @@ public static class Builder {
private String bootstrapServers;
private boolean startConsumerThread;
private Map<String, Object> overrideRemoteLogMetadataManagerProps = Collections.emptyMap();
private Set<TopicIdPartition> topicIdPartitions = Collections.emptySet();
private Supplier<RemotePartitionMetadataStore> remotePartitionMetadataStore = RemotePartitionMetadataStore::new;
private Function<Integer, RemoteLogMetadataTopicPartitioner> remoteLogMetadataTopicPartitioner = RemoteLogMetadataTopicPartitioner::new;

Expand Down Expand Up @@ -85,11 +78,6 @@ public Builder overrideRemoteLogMetadataManagerProps(Map<String, Object> overrid
return this;
}

public Builder topicIdPartitions(Set<TopicIdPartition> topicIdPartitions) {
this.topicIdPartitions = Objects.requireNonNull(topicIdPartitions);
return this;
}

public TopicBasedRemoteLogMetadataManager build() {
Objects.requireNonNull(bootstrapServers);
String logDir = TestUtils.tempDirectory("rlmm_segs_").getAbsolutePath();
Expand All @@ -105,19 +93,12 @@ public TopicBasedRemoteLogMetadataManager build() {
configs.put(REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP, METADATA_TOPIC_PARTITIONS_COUNT);
configs.put(REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR_PROP, METADATA_TOPIC_REPLICATION_FACTOR);
configs.put(REMOTE_LOG_METADATA_TOPIC_RETENTION_MS_PROP, METADATA_TOPIC_RETENTION_MS);

log.debug("TopicBasedRemoteLogMetadataManager configs before adding overridden properties: {}", configs);
// Add override properties.
configs.putAll(overrideRemoteLogMetadataManagerProps);
log.debug("TopicBasedRemoteLogMetadataManager configs after adding overridden properties: {}", configs);

topicBasedRemoteLogMetadataManager.configure(configs);

Assertions.assertDoesNotThrow(() -> TestUtils.waitForCondition(topicBasedRemoteLogMetadataManager::isInitialized, 60_000L,
assertDoesNotThrow(() -> TestUtils.waitForCondition(topicBasedRemoteLogMetadataManager::isInitialized, 60_000L,
"Time out reached before it is initialized successfully"));

topicBasedRemoteLogMetadataManager.onPartitionLeadershipChanges(topicIdPartitions, Collections.emptySet());

return topicBasedRemoteLogMetadataManager;
}
}
Expand Down
63 changes: 0 additions & 63 deletions ...org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleManager.java
View file
Open in desktop

This file was deleted.

Loading