Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
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
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

This file was deleted.

Loading