You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by sa...@apache.org on 2024/06/11 07:12:48 UTC
(kafka) branch 3.8 updated (facbab272fb -> d94a28b4a4f)
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a change to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
from facbab272fb KAFKA-9228: Restart tasks on runtime-only connector config changes (#16053)
new 68d92a5b43f KAFKA-16866 Used the right constant in RemoteLogManagerTest#testFetchQuotaManagerConfig (#16152)
new c6f0db3c606 KAFKA-16785 Migrate TopicBasedRemoteLogMetadataManagerRestartTest to new test infra (#16170)
new 2273e061389 MINOR: Fix missing wait topic finished in TopicBasedRemoteLogMetadataManagerRestartTest (#16171)
new f9c37032ffc KAFKA-16859 Cleanup check if tiered storage is enabled (#16153)
new e4a3da6b099 KAFKA-16852 Adding two thread pools kafka-16852 (#16154)
new 944f4699a7d KAFKA-16880 Update equals and hashcode methods for two attributes (#16173)
new 69158f67f84 KAFKA-16882 Migrate RemoteLogSegmentLifecycleTest to ClusterInstance infra (#16180)
new b6848d699da KAFKA-15776: Introduce remote.fetch.max.timeout.ms to configure DelayedRemoteFetch timeout (#14778)
new 025e791d0cd MINOR: Cleanup the storage module unit tests (#16202)
new 9460e6b266a KAFKA-16884 Refactor RemoteLogManagerConfig with AbstractConfig (#16199)
new 781b93b00d9 KAFKA-16885 Renamed the enableRemoteStorageSystem to isRemoteStorageSystemEnabled (#16256)
new d94a28b4a4f KAFKA-15776: Support added to update remote.fetch.max.wait.ms dynamically (#16203)
The 12 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
Summary of changes:
.../kafka/clients/consumer/ConsumerConfig.java | 5 +-
.../java/kafka/log/remote/RemoteLogManager.java | 2 +-
.../kafka/server/builders/KafkaApisBuilder.java | 2 +-
.../server/builders/ReplicaManagerBuilder.java | 2 +-
core/src/main/scala/kafka/log/LogManager.scala | 4 +-
.../src/main/scala/kafka/server/BrokerServer.scala | 4 +-
.../main/scala/kafka/server/ConfigHandler.scala | 2 +-
.../server/ControllerConfigurationValidator.scala | 3 +-
.../scala/kafka/server/DelayedRemoteFetch.scala | 3 +-
.../scala/kafka/server/DynamicBrokerConfig.scala | 3 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 10 +-
core/src/main/scala/kafka/server/KafkaServer.scala | 4 +-
.../main/scala/kafka/server/ReplicaManager.scala | 4 +-
core/src/main/scala/kafka/zk/AdminZkClient.scala | 4 +-
.../kafka/log/remote/RemoteLogManagerTest.java | 15 +-
.../kafka/server/DelayedRemoteFetchTest.scala | 21 +-
.../test/scala/unit/kafka/log/LogConfigTest.scala | 26 +-
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 2 +-
.../kafka/server/DynamicBrokerConfigTest.scala | 33 ++
.../unit/kafka/server/ReplicaManagerTest.scala | 14 +-
.../log/remote/storage/RemoteLogManagerConfig.java | 558 ++++++++------------
.../storage/RemoteLogMetadataFormatterTest.java | 18 +-
.../storage/RemoteLogMetadataManagerTestUtils.java | 23 +-
.../storage/RemoteLogMetadataSerdeTest.java | 38 +-
.../storage/RemoteLogMetadataTransformTest.java | 40 +-
.../storage/RemoteLogSegmentLifecycleManager.java | 63 ---
.../storage/RemoteLogSegmentLifecycleTest.java | 562 ++++++++-------------
...picBasedRemoteLogMetadataManagerConfigTest.java | 53 +-
.../TopicBasedRemoteLogMetadataManagerHarness.java | 95 ----
...ogMetadataManagerMultipleSubscriptionsTest.java | 41 +-
...icBasedRemoteLogMetadataManagerRestartTest.java | 181 +++----
.../TopicBasedRemoteLogMetadataManagerTest.java | 56 +-
...RemoteLogMetadataManagerWrapperWithHarness.java | 105 ----
.../remote/storage/RemoteLogManagerConfigTest.java | 107 ++--
.../storage/RemoteLogMetadataManagerTest.java | 83 +--
35 files changed, 812 insertions(+), 1374 deletions(-)
delete mode 100644 storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleManager.java
delete mode 100644 storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
delete mode 100644 storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerWrapperWithHarness.java
(kafka) 09/12: MINOR: Cleanup the storage module unit tests (#16202)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 025e791d0cdfc025fd064379f8ebb8b9fe380a26
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Thu Jun 6 21:26:08 2024 +0530
MINOR: Cleanup the storage module unit tests (#16202)
- Use SystemTime instead of MockTime when time is not mocked
- Use static assertions to reduce the line length
- Fold the lines if it exceeds the limit
- rename tp0 to tpId0 when it refers to TopicIdPartition
Reviewers: Kuan-Po (Cooper) Tseng <br...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../storage/RemoteLogMetadataFormatterTest.java | 18 ++---
.../storage/RemoteLogMetadataSerdeTest.java | 38 +++++-----
.../storage/RemoteLogMetadataTransformTest.java | 40 +++++------
...picBasedRemoteLogMetadataManagerConfigTest.java | 53 +++++---------
...ogMetadataManagerMultipleSubscriptionsTest.java | 41 +++++------
...icBasedRemoteLogMetadataManagerRestartTest.java | 25 +++----
.../TopicBasedRemoteLogMetadataManagerTest.java | 56 ++++++++-------
.../storage/RemoteLogMetadataManagerTest.java | 83 ++++++++++++----------
8 files changed, 169 insertions(+), 185 deletions(-)
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataFormatterTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataFormatterTest.java
index e3d1a2aee0c..1380a735fba 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataFormatterTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataFormatterTest.java
@@ -24,7 +24,6 @@ import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteL
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata.CustomMetadata;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
import org.junit.jupiter.api.Test;
import java.io.ByteArrayOutputStream;
@@ -35,6 +34,7 @@ import java.util.Map;
import java.util.Optional;
import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.COPY_SEGMENT_STARTED;
public class RemoteLogMetadataFormatterTest {
private static final Uuid TOPIC_ID = Uuid.randomUuid();
@@ -51,12 +51,12 @@ public class RemoteLogMetadataFormatterTest {
RemoteLogSegmentId remoteLogSegmentId = new RemoteLogSegmentId(TP0, SEGMENT_ID);
Optional<CustomMetadata> customMetadata = Optional.of(new CustomMetadata(new byte[10]));
RemoteLogSegmentMetadata remoteLogMetadata = new RemoteLogSegmentMetadata(
- remoteLogSegmentId, 0L, 100L, -1L, 1,
- 123L, 1024, customMetadata,
- RemoteLogSegmentState.COPY_SEGMENT_STARTED, segLeaderEpochs);
+ remoteLogSegmentId, 0L, 100L, -1L, 1, 123L, 1024, customMetadata, COPY_SEGMENT_STARTED,
+ segLeaderEpochs);
byte[] metadataBytes = new RemoteLogMetadataSerde().serialize(remoteLogMetadata);
- ConsumerRecord<byte[], byte[]> metadataRecord = new ConsumerRecord<>("__remote_log_metadata", 0, 0, null, metadataBytes);
+ ConsumerRecord<byte[], byte[]> metadataRecord = new ConsumerRecord<>(
+ "__remote_log_metadata", 0, 0, null, metadataBytes);
String expected = String.format(
"partition: 0, offset: 0, value: " +
@@ -68,9 +68,11 @@ public class RemoteLogMetadataFormatterTest {
TOPIC_ID, SEGMENT_ID);
try (ByteArrayOutputStream baos = new ByteArrayOutputStream();
PrintStream ps = new PrintStream(baos)) {
- RemoteLogMetadataSerde.RemoteLogMetadataFormatter formatter = new RemoteLogMetadataSerde.RemoteLogMetadataFormatter();
- formatter.writeTo(metadataRecord, ps);
- assertEquals(expected, baos.toString());
+ try (RemoteLogMetadataSerde.RemoteLogMetadataFormatter formatter =
+ new RemoteLogMetadataSerde.RemoteLogMetadataFormatter()) {
+ formatter.writeTo(metadataRecord, ps);
+ assertEquals(expected, baos.toString());
+ }
}
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataSerdeTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataSerdeTest.java
index 5b48790c7fd..b1b91dacf23 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataSerdeTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataSerdeTest.java
@@ -19,48 +19,48 @@ package org.apache.kafka.server.log.remote.metadata.storage;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogMetadataSerde;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata.CustomMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
import org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteMetadata;
-import org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState;
import org.apache.kafka.server.log.remote.storage.RemoteLogMetadata;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.COPY_SEGMENT_STARTED;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.COPY_SEGMENT_FINISHED;
+import static org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState.DELETE_PARTITION_MARKED;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
public class RemoteLogMetadataSerdeTest {
public static final String TOPIC = "foo";
private static final TopicIdPartition TP0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(TOPIC, 0));
- private final Time time = new MockTime(1);
+ private final Time time = new SystemTime();
@Test
public void testRemoteLogSegmentMetadataSerde() {
RemoteLogSegmentMetadata remoteLogSegmentMetadata = createRemoteLogSegmentMetadata();
-
doTestRemoteLogMetadataSerde(remoteLogSegmentMetadata);
}
@Test
public void testRemoteLogSegmentMetadataUpdateSerde() {
RemoteLogSegmentMetadataUpdate remoteLogSegmentMetadataUpdate = createRemoteLogSegmentMetadataUpdate();
-
doTestRemoteLogMetadataSerde(remoteLogSegmentMetadataUpdate);
}
@Test
public void testRemotePartitionDeleteMetadataSerde() {
RemotePartitionDeleteMetadata remotePartitionDeleteMetadata = createRemotePartitionDeleteMetadata();
-
doTestRemoteLogMetadataSerde(remotePartitionDeleteMetadata);
}
@@ -70,24 +70,19 @@ public class RemoteLogMetadataSerdeTest {
segLeaderEpochs.put(1, 20L);
segLeaderEpochs.put(2, 80L);
RemoteLogSegmentId remoteLogSegmentId = new RemoteLogSegmentId(TP0, Uuid.randomUuid());
- return new RemoteLogSegmentMetadata(remoteLogSegmentId, 0L, 100L, -1L, 1,
- time.milliseconds(), 1024,
- Optional.of(new CustomMetadata(new byte[] {0, 1, 2, 3})),
- RemoteLogSegmentState.COPY_SEGMENT_STARTED,
- segLeaderEpochs
- );
+ return new RemoteLogSegmentMetadata(remoteLogSegmentId, 0L, 100L, -1L, 1,
+ time.milliseconds(), 1024, Optional.of(new CustomMetadata(new byte[] {0, 1, 2, 3})),
+ COPY_SEGMENT_STARTED, segLeaderEpochs);
}
private RemoteLogSegmentMetadataUpdate createRemoteLogSegmentMetadataUpdate() {
RemoteLogSegmentId remoteLogSegmentId = new RemoteLogSegmentId(TP0, Uuid.randomUuid());
return new RemoteLogSegmentMetadataUpdate(remoteLogSegmentId, time.milliseconds(),
- Optional.of(new CustomMetadata(new byte[] {0, 1, 2, 3})),
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED, 2);
+ Optional.of(new CustomMetadata(new byte[] {0, 1, 2, 3})), COPY_SEGMENT_FINISHED, 2);
}
private RemotePartitionDeleteMetadata createRemotePartitionDeleteMetadata() {
- return new RemotePartitionDeleteMetadata(TP0, RemotePartitionDeleteState.DELETE_PARTITION_MARKED,
- time.milliseconds(), 0);
+ return new RemotePartitionDeleteMetadata(TP0, DELETE_PARTITION_MARKED, time.milliseconds(), 0);
}
private void doTestRemoteLogMetadataSerde(RemoteLogMetadata remoteLogMetadata) {
@@ -96,16 +91,17 @@ public class RemoteLogMetadataSerdeTest {
byte[] metadataBytes = serializer.serialize(remoteLogMetadata);
// Deserialize the bytes and check the RemoteLogMetadata object is as expected.
- // Created another RemoteLogMetadataSerde instance to depict the real usecase of serializer and deserializer having their own instances.
+ // Created another RemoteLogMetadataSerde instance to depict the real usecase of serializer and
+ // deserializer having their own instances.
RemoteLogMetadataSerde deserializer = new RemoteLogMetadataSerde();
RemoteLogMetadata deserializedRemoteLogMetadata = deserializer.deserialize(metadataBytes);
- Assertions.assertEquals(remoteLogMetadata, deserializedRemoteLogMetadata);
+ assertEquals(remoteLogMetadata, deserializedRemoteLogMetadata);
}
@Test
public void testInvalidRemoteStorageMetadata() {
// Serializing receives an exception as it does not have the expected RemoteLogMetadata registered in serdes.
- Assertions.assertThrows(IllegalArgumentException.class,
+ assertThrows(IllegalArgumentException.class,
() -> new RemoteLogMetadataSerde().serialize(new InvalidRemoteLogMetadata(1, time.milliseconds())));
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataTransformTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataTransformTest.java
index 504f47e17a5..70770542eb9 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataTransformTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataTransformTest.java
@@ -19,7 +19,7 @@ package org.apache.kafka.server.log.remote.metadata.storage;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.common.ApiMessageAndVersion;
import org.apache.kafka.server.log.remote.metadata.storage.serialization.RemoteLogSegmentMetadataTransform;
@@ -29,60 +29,56 @@ import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata.CustomMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
import org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteMetadata;
-import org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
import java.util.Collections;
import java.util.Optional;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.COPY_SEGMENT_FINISHED;
+import static org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState.DELETE_PARTITION_STARTED;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+
public class RemoteLogMetadataTransformTest {
private static final TopicIdPartition TP0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
- private final Time time = new MockTime(1);
+ private final Time time = new SystemTime();
@Test
public void testRemoteLogSegmentMetadataTransform() {
RemoteLogSegmentMetadataTransform metadataTransform = new RemoteLogSegmentMetadataTransform();
-
RemoteLogSegmentMetadata metadata = createRemoteLogSegmentMetadata();
ApiMessageAndVersion apiMessageAndVersion = metadataTransform.toApiMessageAndVersion(metadata);
RemoteLogSegmentMetadata remoteLogSegmentMetadataFromRecord = metadataTransform
.fromApiMessageAndVersion(apiMessageAndVersion);
-
- Assertions.assertEquals(metadata, remoteLogSegmentMetadataFromRecord);
+ assertEquals(metadata, remoteLogSegmentMetadataFromRecord);
}
@Test
public void testRemoteLogSegmentMetadataUpdateTransform() {
RemoteLogSegmentMetadataUpdateTransform metadataUpdateTransform = new RemoteLogSegmentMetadataUpdateTransform();
-
- RemoteLogSegmentMetadataUpdate metadataUpdate =
- new RemoteLogSegmentMetadataUpdate(new RemoteLogSegmentId(TP0, Uuid.randomUuid()), time.milliseconds(),
- Optional.of(new CustomMetadata(new byte[]{0, 1, 2, 3})),
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED, 1);
+ RemoteLogSegmentMetadataUpdate metadataUpdate = new RemoteLogSegmentMetadataUpdate(
+ new RemoteLogSegmentId(TP0, Uuid.randomUuid()), time.milliseconds(),
+ Optional.of(new CustomMetadata(new byte[]{0, 1, 2, 3})), COPY_SEGMENT_FINISHED, 1);
ApiMessageAndVersion apiMessageAndVersion = metadataUpdateTransform.toApiMessageAndVersion(metadataUpdate);
- RemoteLogSegmentMetadataUpdate metadataUpdateFromRecord = metadataUpdateTransform.fromApiMessageAndVersion(apiMessageAndVersion);
-
- Assertions.assertEquals(metadataUpdate, metadataUpdateFromRecord);
+ RemoteLogSegmentMetadataUpdate metadataUpdateFromRecord =
+ metadataUpdateTransform.fromApiMessageAndVersion(apiMessageAndVersion);
+ assertEquals(metadataUpdate, metadataUpdateFromRecord);
}
private RemoteLogSegmentMetadata createRemoteLogSegmentMetadata() {
RemoteLogSegmentId remoteLogSegmentId = new RemoteLogSegmentId(TP0, Uuid.randomUuid());
return new RemoteLogSegmentMetadata(remoteLogSegmentId, 0L, 100L, -1L, 1,
- time.milliseconds(), 1024, Collections.singletonMap(0, 0L));
+ time.milliseconds(), 1024, Collections.singletonMap(0, 0L));
}
@Test
public void testRemoteLogPartitionMetadataTransform() {
RemotePartitionDeleteMetadataTransform transform = new RemotePartitionDeleteMetadataTransform();
-
RemotePartitionDeleteMetadata partitionDeleteMetadata
- = new RemotePartitionDeleteMetadata(TP0, RemotePartitionDeleteState.DELETE_PARTITION_STARTED, time.milliseconds(), 1);
+ = new RemotePartitionDeleteMetadata(TP0, DELETE_PARTITION_STARTED, time.milliseconds(), 1);
ApiMessageAndVersion apiMessageAndVersion = transform.toApiMessageAndVersion(partitionDeleteMetadata);
- RemotePartitionDeleteMetadata partitionDeleteMetadataFromRecord = transform.fromApiMessageAndVersion(apiMessageAndVersion);
-
- Assertions.assertEquals(partitionDeleteMetadata, partitionDeleteMetadataFromRecord);
+ RemotePartitionDeleteMetadata partitionDeleteMetadataFromRecord =
+ transform.fromApiMessageAndVersion(apiMessageAndVersion);
+ assertEquals(partitionDeleteMetadata, partitionDeleteMetadataFromRecord);
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerConfigTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerConfigTest.java
index 8e3985d0d5f..34f1fb08366 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerConfigTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerConfigTest.java
@@ -20,10 +20,7 @@ import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.test.TestUtils;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Test;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
import java.util.AbstractMap;
import java.util.HashMap;
@@ -38,10 +35,10 @@ import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemo
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;
-public class TopicBasedRemoteLogMetadataManagerConfigTest {
- private static final Logger log = LoggerFactory.getLogger(TopicBasedRemoteLogMetadataManagerConfigTest.class);
+import static org.junit.jupiter.api.Assertions.assertEquals;
- private static final String BOOTSTRAP_SERVERS = "localhost:9091";
+public class TopicBasedRemoteLogMetadataManagerConfigTest {
+ private static final String BOOTSTRAP_SERVERS = "localhost:2222";
@Test
public void testValidConfig() {
@@ -60,41 +57,32 @@ public class TopicBasedRemoteLogMetadataManagerConfigTest {
// Check for topic properties
TopicBasedRemoteLogMetadataManagerConfig rlmmConfig = new TopicBasedRemoteLogMetadataManagerConfig(props);
- Assertions.assertEquals(props.get(REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP), rlmmConfig.metadataTopicPartitionsCount());
+ assertEquals(props.get(REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP), rlmmConfig.metadataTopicPartitionsCount());
// Check for common client configs.
- Assertions.assertEquals(BOOTSTRAP_SERVERS, rlmmConfig.commonProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
- Assertions.assertEquals(BOOTSTRAP_SERVERS, rlmmConfig.producerProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
- Assertions.assertEquals(BOOTSTRAP_SERVERS, rlmmConfig.consumerProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
+ assertEquals(BOOTSTRAP_SERVERS, rlmmConfig.commonProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
+ assertEquals(BOOTSTRAP_SERVERS, rlmmConfig.producerProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
+ assertEquals(BOOTSTRAP_SERVERS, rlmmConfig.consumerProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG));
for (Map.Entry<String, Object> entry : commonClientConfig.entrySet()) {
- log.info("Checking config: " + entry.getKey());
- Assertions.assertEquals(entry.getValue(),
- rlmmConfig.commonProperties().get(entry.getKey()));
- Assertions.assertEquals(entry.getValue(),
- rlmmConfig.producerProperties().get(entry.getKey()));
- Assertions.assertEquals(entry.getValue(),
- rlmmConfig.consumerProperties().get(entry.getKey()));
+ assertEquals(entry.getValue(), rlmmConfig.commonProperties().get(entry.getKey()));
+ assertEquals(entry.getValue(), rlmmConfig.producerProperties().get(entry.getKey()));
+ assertEquals(entry.getValue(), rlmmConfig.consumerProperties().get(entry.getKey()));
}
-
// Check for producer configs.
for (Map.Entry<String, Object> entry : producerConfig.entrySet()) {
- log.info("Checking config: " + entry.getKey());
- Assertions.assertEquals(entry.getValue(),
- rlmmConfig.producerProperties().get(entry.getKey()));
+ assertEquals(entry.getValue(), rlmmConfig.producerProperties().get(entry.getKey()));
}
-
// Check for consumer configs.
for (Map.Entry<String, Object> entry : consumerConfig.entrySet()) {
- log.info("Checking config: " + entry.getKey());
- Assertions.assertEquals(entry.getValue(),
- rlmmConfig.consumerProperties().get(entry.getKey()));
+ assertEquals(entry.getValue(), rlmmConfig.consumerProperties().get(entry.getKey()));
}
}
@Test
public void testCommonProducerConsumerOverridesConfig() {
- Map.Entry<String, Long> overrideEntry = new AbstractMap.SimpleImmutableEntry<>(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, 60000L);
+ Map.Entry<String, Long> overrideEntry =
+ new AbstractMap.SimpleImmutableEntry<>(CommonClientConfigs.METADATA_MAX_AGE_CONFIG, 60000L);
Map<String, Object> commonClientConfig = new HashMap<>();
commonClientConfig.put(CommonClientConfigs.RETRIES_CONFIG, 10);
commonClientConfig.put(CommonClientConfigs.RETRY_BACKOFF_MS_CONFIG, 1000L);
@@ -114,12 +102,9 @@ public class TopicBasedRemoteLogMetadataManagerConfigTest {
Map<String, Object> props = createValidConfigProps(commonClientConfig, producerConfig, consumerConfig);
TopicBasedRemoteLogMetadataManagerConfig rlmmConfig = new TopicBasedRemoteLogMetadataManagerConfig(props);
- Assertions.assertEquals(overrideCommonPropValue,
- rlmmConfig.commonProperties().get(overrideEntry.getKey()));
- Assertions.assertEquals(overriddenProducerPropValue,
- rlmmConfig.producerProperties().get(overrideEntry.getKey()));
- Assertions.assertEquals(overriddenConsumerPropValue,
- rlmmConfig.consumerProperties().get(overrideEntry.getKey()));
+ assertEquals(overrideCommonPropValue, rlmmConfig.commonProperties().get(overrideEntry.getKey()));
+ assertEquals(overriddenProducerPropValue, rlmmConfig.producerProperties().get(overrideEntry.getKey()));
+ assertEquals(overriddenConsumerPropValue, rlmmConfig.consumerProperties().get(overrideEntry.getKey()));
}
private Map<String, Object> createValidConfigProps(Map<String, Object> commonClientConfig,
@@ -129,7 +114,6 @@ public class TopicBasedRemoteLogMetadataManagerConfigTest {
props.put(REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX + CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, BOOTSTRAP_SERVERS);
props.put(BROKER_ID, 1);
props.put(LOG_DIR, TestUtils.tempDirectory().getAbsolutePath());
-
props.put(REMOTE_LOG_METADATA_TOPIC_REPLICATION_FACTOR_PROP, (short) 3);
props.put(REMOTE_LOG_METADATA_TOPIC_PARTITIONS_PROP, 10);
props.put(REMOTE_LOG_METADATA_TOPIC_RETENTION_MS_PROP, 60 * 60 * 1000L);
@@ -138,17 +122,14 @@ public class TopicBasedRemoteLogMetadataManagerConfigTest {
for (Map.Entry<String, Object> entry : commonClientConfig.entrySet()) {
props.put(REMOTE_LOG_METADATA_COMMON_CLIENT_PREFIX + entry.getKey(), entry.getValue());
}
-
// producer configs
for (Map.Entry<String, Object> entry : producerConfig.entrySet()) {
props.put(REMOTE_LOG_METADATA_PRODUCER_PREFIX + entry.getKey(), entry.getValue());
}
-
//consumer configs
for (Map.Entry<String, Object> entry : consumerConfig.entrySet()) {
props.put(REMOTE_LOG_METADATA_CONSUMER_PREFIX + entry.getKey(), entry.getValue());
}
-
return props;
}
}
\ No newline at end of file
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java
index 916b475c143..d02da098040 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest.java
@@ -27,12 +27,11 @@ import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -44,6 +43,10 @@ import java.util.concurrent.ExecutionException;
import java.util.concurrent.Phaser;
import java.util.concurrent.TimeUnit;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.clearInvocations;
import static org.mockito.Mockito.doAnswer;
@@ -55,10 +58,7 @@ import static org.mockito.Mockito.verify;
@ClusterTestDefaults(brokers = 3)
public class TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest {
private final ClusterInstance clusterInstance;
-
- private static final int SEG_SIZE = 1024 * 1024;
-
- private final Time time = new MockTime(1);
+ private final Time time = new SystemTime();
TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest(ClusterInstance clusterInstance) {
this.clusterInstance = clusterInstance;
@@ -125,24 +125,25 @@ public class TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest {
// Add segments for these partitions but an exception is received as they have not yet been subscribed.
// These messages would have been published to the respective metadata topic partitions but the ConsumerManager
// has not yet been subscribing as they are not yet registered.
+ int segSize = 1048576;
RemoteLogSegmentMetadata leaderSegmentMetadata = new RemoteLogSegmentMetadata(new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- ExecutionException exception = Assertions.assertThrows(ExecutionException.class, () -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(leaderSegmentMetadata).get());
- Assertions.assertEquals("org.apache.kafka.common.KafkaException: This consumer is not assigned to the target partition 0. Currently assigned partitions: []",
+ time.milliseconds(), segSize, Collections.singletonMap(0, 0L));
+ ExecutionException exception = assertThrows(ExecutionException.class,
+ () -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(leaderSegmentMetadata).get());
+ assertEquals("org.apache.kafka.common.KafkaException: This consumer is not assigned to the target partition 0. Currently assigned partitions: []",
exception.getMessage());
RemoteLogSegmentMetadata followerSegmentMetadata = new RemoteLogSegmentMetadata(new RemoteLogSegmentId(followerTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- exception = Assertions.assertThrows(ExecutionException.class, () -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(followerSegmentMetadata).get());
- Assertions.assertEquals("org.apache.kafka.common.KafkaException: This consumer is not assigned to the target partition 0. Currently assigned partitions: []",
+ time.milliseconds(), segSize, Collections.singletonMap(0, 0L));
+ exception = assertThrows(ExecutionException.class, () -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(followerSegmentMetadata).get());
+ assertEquals("org.apache.kafka.common.KafkaException: This consumer is not assigned to the target partition 0. Currently assigned partitions: []",
exception.getMessage());
// `listRemoteLogSegments` will receive an exception as these topic partitions are not yet registered.
- Assertions.assertThrows(RemoteStorageException.class, () -> remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition));
- Assertions.assertThrows(RemoteStorageException.class, () -> remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition));
-
+ assertThrows(RemoteStorageException.class, () -> remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition));
+ assertThrows(RemoteStorageException.class, () -> remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition));
remoteLogMetadataManager.onPartitionLeadershipChanges(Collections.singleton(leaderTopicIdPartition),
Collections.emptySet());
@@ -156,8 +157,8 @@ public class TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest {
// leader partitions would have received as it is registered, but follower partition is not yet registered,
// hence it throws an exception.
- Assertions.assertTrue(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition).hasNext());
- Assertions.assertThrows(RemoteStorageException.class, () -> remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition));
+ assertTrue(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition).hasNext());
+ assertThrows(RemoteStorageException.class, () -> remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition));
// Register follower partition
// Phaser::bulkRegister and Phaser::register provide the "countUp" feature
@@ -172,15 +173,15 @@ public class TopicBasedRemoteLogMetadataManagerMultipleSubscriptionsTest {
verify(spyRemotePartitionMetadataStore).markInitialized(followerTopicIdPartition);
verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadata(followerSegmentMetadata);
// In this state, all the metadata should be available in RLMM for both leader and follower partitions.
- Assertions.assertTrue(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition).hasNext(), "No segments found");
- Assertions.assertTrue(remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition).hasNext(), "No segments found");
+ assertTrue(remoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition).hasNext(), "No segments found");
+ assertTrue(remoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition).hasNext(), "No segments found");
}
}
private void createTopic(String topic, Map<Integer, List<Integer>> replicasAssignments) {
try (Admin admin = Admin.create(Collections.singletonMap(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG, clusterInstance.bootstrapServers()))) {
admin.createTopics(Collections.singletonList(new NewTopic(topic, replicasAssignments)));
- Assertions.assertDoesNotThrow(() -> clusterInstance.waitForTopic(topic, replicasAssignments.size()));
+ assertDoesNotThrow(() -> clusterInstance.waitForTopic(topic, replicasAssignments.size()));
}
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
index dee686dc036..a08f12d9ae2 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
@@ -24,12 +24,11 @@ import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.test.TestUtils;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.extension.ExtendWith;
@@ -37,14 +36,13 @@ import java.util.Arrays;
import java.util.Collections;
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.LOG_DIR;
+import static org.junit.jupiter.api.Assertions.assertTrue;
@ExtendWith(value = ClusterTestExtensions.class)
@Tag("integration")
public class TopicBasedRemoteLogMetadataManagerRestartTest {
- private static final int SEG_SIZE = 1024 * 1024;
-
- private final Time time = new MockTime(1);
+ private final Time time = new SystemTime();
private final String logDir = TestUtils.tempDirectory("_rlmm_segs_").getAbsolutePath();
private final ClusterInstance clusterInstance;
@@ -76,16 +74,17 @@ public class TopicBasedRemoteLogMetadataManagerRestartTest {
clusterInstance.waitForTopic(leaderTopic, 1);
clusterInstance.waitForTopic(followerTopic, 1);
- final TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
- final TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));
+ TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
+ TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));
+ int segSize = 1048576;
RemoteLogSegmentMetadata leaderSegmentMetadata = new RemoteLogSegmentMetadata(
new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
+ time.milliseconds(), segSize, Collections.singletonMap(0, 0L));
RemoteLogSegmentMetadata followerSegmentMetadata = new RemoteLogSegmentMetadata(
new RemoteLogSegmentId(followerTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
+ time.milliseconds(), segSize, Collections.singletonMap(0, 0L));
try (TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager = createTopicBasedRemoteLogMetadataManager()) {
// Register these partitions to RemoteLogMetadataManager.
@@ -115,12 +114,14 @@ public class TopicBasedRemoteLogMetadataManagerRestartTest {
RemoteLogSegmentMetadata leaderSegmentMetadata2 = new RemoteLogSegmentMetadata(
new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()),
101, 200, -1L, 0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 101L));
+ time.milliseconds(), segSize, Collections.singletonMap(0, 101L));
topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(leaderSegmentMetadata2).get();
// Check that both the stored segment and recently added segment are available.
- Assertions.assertTrue(TestUtils.sameElementsWithoutOrder(Arrays.asList(leaderSegmentMetadata, leaderSegmentMetadata2).iterator(),
- topicBasedRemoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition)));
+ assertTrue(TestUtils.sameElementsWithoutOrder(
+ Arrays.asList(leaderSegmentMetadata, leaderSegmentMetadata2).iterator(),
+ topicBasedRemoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition))
+ );
}
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java
index 183a8493412..8b9cfd0700c 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerTest.java
@@ -25,14 +25,13 @@ import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteResourceNotFoundException;
import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
import org.junit.jupiter.api.AfterEach;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.extension.ExtendWith;
import java.io.IOException;
@@ -42,6 +41,10 @@ import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeUnit;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+import static org.junit.jupiter.api.Assertions.assertTrue;
import static org.mockito.ArgumentMatchers.any;
import static org.mockito.Mockito.doAnswer;
import static org.mockito.Mockito.spy;
@@ -50,11 +53,10 @@ import static org.mockito.Mockito.verify;
@ExtendWith(ClusterTestExtensions.class)
@ClusterTestDefaults(brokers = 3)
public class TopicBasedRemoteLogMetadataManagerTest {
- private static final int SEG_SIZE = 1024 * 1024;
-
+ private static final int SEG_SIZE = 1048576;
private final ClusterInstance clusterInstance;
private final RemotePartitionMetadataStore spyRemotePartitionMetadataEventHandler = spy(new RemotePartitionMetadataStore());
- private final Time time = new MockTime(1);
+ private final Time time = new SystemTime();
private TopicBasedRemoteLogMetadataManager remoteLogMetadataManager;
TopicBasedRemoteLogMetadataManagerTest(ClusterInstance clusterInstance) {
@@ -83,7 +85,7 @@ public class TopicBasedRemoteLogMetadataManagerTest {
admin.createTopics(Collections.singletonList(new NewTopic(topic, 1, (short) 1))).all().get();
clusterInstance.waitForTopic(topic, 1);
boolean doesTopicExist = topicBasedRlmm().doesTopicExist(admin, topic);
- Assertions.assertTrue(doesTopicExist);
+ assertTrue(doesTopicExist);
}
}
@@ -92,7 +94,7 @@ public class TopicBasedRemoteLogMetadataManagerTest {
try (Admin admin = clusterInstance.createAdminClient()) {
String topic = "dummy-test-topic";
boolean doesTopicExist = topicBasedRlmm().doesTopicExist(admin, topic);
- Assertions.assertFalse(doesTopicExist);
+ assertFalse(doesTopicExist);
}
}
@@ -139,37 +141,37 @@ public class TopicBasedRemoteLogMetadataManagerTest {
RemoteLogSegmentMetadata leaderSegmentMetadata = new RemoteLogSegmentMetadata(new RemoteLogSegmentId(newLeaderTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- Assertions.assertThrows(Exception.class, () -> topicBasedRlmm().addRemoteLogSegmentMetadata(leaderSegmentMetadata).get());
+ assertThrows(Exception.class, () -> topicBasedRlmm().addRemoteLogSegmentMetadata(leaderSegmentMetadata).get());
RemoteLogSegmentMetadata followerSegmentMetadata = new RemoteLogSegmentMetadata(new RemoteLogSegmentId(newFollowerTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- Assertions.assertThrows(Exception.class, () -> topicBasedRlmm().addRemoteLogSegmentMetadata(followerSegmentMetadata).get());
+ assertThrows(Exception.class, () -> topicBasedRlmm().addRemoteLogSegmentMetadata(followerSegmentMetadata).get());
// `listRemoteLogSegments` will receive an exception as these topic partitions are not yet registered.
- Assertions.assertThrows(RemoteResourceNotFoundException.class, () -> topicBasedRlmm().listRemoteLogSegments(newLeaderTopicIdPartition));
- Assertions.assertThrows(RemoteResourceNotFoundException.class, () -> topicBasedRlmm().listRemoteLogSegments(newFollowerTopicIdPartition));
+ assertThrows(RemoteResourceNotFoundException.class, () -> topicBasedRlmm().listRemoteLogSegments(newLeaderTopicIdPartition));
+ assertThrows(RemoteResourceNotFoundException.class, () -> topicBasedRlmm().listRemoteLogSegments(newFollowerTopicIdPartition));
topicBasedRlmm().onPartitionLeadershipChanges(Collections.singleton(newLeaderTopicIdPartition),
Collections.singleton(newFollowerTopicIdPartition));
// RemoteLogSegmentMetadata events are already published, and topicBasedRlmm's consumer manager will start
// fetching those events and build the cache.
- Assertions.assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
- Assertions.assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
verify(spyRemotePartitionMetadataEventHandler).markInitialized(newLeaderTopicIdPartition);
verify(spyRemotePartitionMetadataEventHandler).markInitialized(newFollowerTopicIdPartition);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(leaderSegmentMetadata);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(followerSegmentMetadata);
- Assertions.assertTrue(topicBasedRlmm().listRemoteLogSegments(newLeaderTopicIdPartition).hasNext());
- Assertions.assertTrue(topicBasedRlmm().listRemoteLogSegments(newFollowerTopicIdPartition).hasNext());
+ assertTrue(topicBasedRlmm().listRemoteLogSegments(newLeaderTopicIdPartition).hasNext());
+ assertTrue(topicBasedRlmm().listRemoteLogSegments(newFollowerTopicIdPartition).hasNext());
}
@ClusterTest
public void testRemoteLogSizeCalculationForUnknownTopicIdPartitionThrows() {
TopicIdPartition topicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("singleton", 0));
- Assertions.assertThrows(RemoteResourceNotFoundException.class, () -> topicBasedRlmm().remoteLogSize(topicIdPartition, 0));
+ assertThrows(RemoteResourceNotFoundException.class, () -> topicBasedRlmm().remoteLogSize(topicIdPartition, 0));
}
@ClusterTest
@@ -206,8 +208,8 @@ public class TopicBasedRemoteLogMetadataManagerTest {
// RemoteLogSegmentMetadata events are already published, and topicBasedRlmm's consumer manager will start
// fetching those events and build the cache.
- Assertions.assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
- Assertions.assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
verify(spyRemotePartitionMetadataEventHandler).markInitialized(topicIdPartition);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(firstSegmentMetadata);
@@ -215,7 +217,7 @@ public class TopicBasedRemoteLogMetadataManagerTest {
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(thirdSegmentMetadata);
Long remoteLogSize = topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 0);
- Assertions.assertEquals(SEG_SIZE * 6, remoteLogSize);
+ assertEquals(SEG_SIZE * 6, remoteLogSize);
}
@ClusterTest
@@ -251,16 +253,16 @@ public class TopicBasedRemoteLogMetadataManagerTest {
// RemoteLogSegmentMetadata events are already published, and topicBasedRlmm's consumer manager will start
// fetching those events and build the cache.
- Assertions.assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
- Assertions.assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
verify(spyRemotePartitionMetadataEventHandler).markInitialized(topicIdPartition);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(firstSegmentMetadata);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(secondSegmentMetadata);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(thirdSegmentMetadata);
- Assertions.assertEquals(SEG_SIZE, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 0));
- Assertions.assertEquals(SEG_SIZE * 2, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 1));
- Assertions.assertEquals(SEG_SIZE * 3, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 2));
+ assertEquals(SEG_SIZE, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 0));
+ assertEquals(SEG_SIZE * 2, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 1));
+ assertEquals(SEG_SIZE * 3, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 2));
}
@ClusterTest
@@ -293,13 +295,13 @@ public class TopicBasedRemoteLogMetadataManagerTest {
// RemoteLogSegmentMetadata events are already published, and topicBasedRlmm's consumer manager will start
// fetching those events and build the cache.
- Assertions.assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
- Assertions.assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(initializationLatch.await(30_000, TimeUnit.MILLISECONDS));
+ assertTrue(handleRemoteLogSegmentMetadataLatch.await(30_000, TimeUnit.MILLISECONDS));
verify(spyRemotePartitionMetadataEventHandler).markInitialized(topicIdPartition);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(firstSegmentMetadata);
verify(spyRemotePartitionMetadataEventHandler).handleRemoteLogSegmentMetadata(secondSegmentMetadata);
- Assertions.assertEquals(0, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 9001));
+ assertEquals(0, topicBasedRemoteLogMetadataManager.remoteLogSize(topicIdPartition, 9001));
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManagerTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManagerTest.java
index b67c316e5e4..d970dd8b68c 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManagerTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogMetadataManagerTest.java
@@ -23,33 +23,39 @@ import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.metadata.storage.RemoteLogMetadataManagerTestUtils;
import org.apache.kafka.server.log.remote.metadata.storage.RemotePartitionMetadataStore;
import org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager;
-import org.junit.jupiter.api.Assertions;
import org.junit.jupiter.api.Tag;
import org.junit.jupiter.api.extension.ExtendWith;
-//import java.io.IOException;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
+import static org.junit.jupiter.api.Assertions.assertDoesNotThrow;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertThrows;
+
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.COPY_SEGMENT_FINISHED;
+import static org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState.DELETE_PARTITION_FINISHED;
+import static org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState.DELETE_PARTITION_MARKED;
+import static org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteState.DELETE_PARTITION_STARTED;
+
@Tag("integration")
@ExtendWith(ClusterTestExtensions.class)
@ClusterTestDefaults(brokers = 3)
public class RemoteLogMetadataManagerTest {
private final ClusterInstance clusterInstance;
-
private static final TopicIdPartition TP0 = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
- private static final int SEG_SIZE = 1024 * 1024;
+ private static final int SEG_SIZE = 1048576;
private static final int BROKER_ID_0 = 0;
private static final int BROKER_ID_1 = 1;
-
- private final Time time = new MockTime(1);
+ private final Time time = new SystemTime();
RemoteLogMetadataManagerTest(ClusterInstance clusterInstance) {
this.clusterInstance = clusterInstance;
@@ -71,26 +77,25 @@ public class RemoteLogMetadataManagerTest {
// 1.Create a segment with state COPY_SEGMENT_STARTED, and this segment should not be available.
Map<Integer, Long> segmentLeaderEpochs = Collections.singletonMap(0, 101L);
RemoteLogSegmentId segmentId = new RemoteLogSegmentId(TP0, Uuid.randomUuid());
- RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(segmentId, 101L, 200L, -1L, BROKER_ID_0,
- time.milliseconds(), SEG_SIZE, segmentLeaderEpochs);
+ RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(
+ segmentId, 101L, 200L, -1L, BROKER_ID_0, time.milliseconds(), SEG_SIZE, segmentLeaderEpochs);
// Wait until the segment is added successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(segmentMetadata).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(segmentMetadata).get());
// Search should not return the above segment.
- Assertions.assertFalse(remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 0, 150).isPresent());
+ assertFalse(remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 0, 150).isPresent());
// 2.Move that segment to COPY_SEGMENT_FINISHED state and this segment should be available.
- RemoteLogSegmentMetadataUpdate segmentMetadataUpdate = new RemoteLogSegmentMetadataUpdate(segmentId, time.milliseconds(),
- Optional.empty(),
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED,
- BROKER_ID_1);
+ RemoteLogSegmentMetadataUpdate segmentMetadataUpdate = new RemoteLogSegmentMetadataUpdate(
+ segmentId, time.milliseconds(), Optional.empty(), COPY_SEGMENT_FINISHED, BROKER_ID_1);
// Wait until the segment is updated successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.updateRemoteLogSegmentMetadata(segmentMetadataUpdate).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.updateRemoteLogSegmentMetadata(segmentMetadataUpdate).get());
RemoteLogSegmentMetadata expectedSegmentMetadata = segmentMetadata.createWithUpdates(segmentMetadataUpdate);
// Search should return the above segment.
- Optional<RemoteLogSegmentMetadata> segmentMetadataForOffset150 = remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 0, 150);
- Assertions.assertEquals(Optional.of(expectedSegmentMetadata), segmentMetadataForOffset150);
+ Optional<RemoteLogSegmentMetadata> segmentMetadataForOffset150 =
+ remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 0, 150);
+ assertEquals(Optional.of(expectedSegmentMetadata), segmentMetadataForOffset150);
}
}
@@ -111,46 +116,46 @@ public class RemoteLogMetadataManagerTest {
segmentLeaderEpochs.put(2, 50L);
segmentLeaderEpochs.put(3, 80L);
RemoteLogSegmentId segmentId = new RemoteLogSegmentId(TP0, Uuid.randomUuid());
- RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(segmentId, 0L, 100L,
- -1L, BROKER_ID_0, time.milliseconds(), SEG_SIZE,
- segmentLeaderEpochs);
+ RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(
+ segmentId, 0L, 100L, -1L, BROKER_ID_0, time.milliseconds(), SEG_SIZE, segmentLeaderEpochs);
// Wait until the segment is added successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(segmentMetadata).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.addRemoteLogSegmentMetadata(segmentMetadata).get());
RemoteLogSegmentMetadataUpdate segmentMetadataUpdate = new RemoteLogSegmentMetadataUpdate(
- segmentId, time.milliseconds(), Optional.empty(), RemoteLogSegmentState.COPY_SEGMENT_FINISHED, BROKER_ID_1);
+ segmentId, time.milliseconds(), Optional.empty(), COPY_SEGMENT_FINISHED, BROKER_ID_1);
// Wait until the segment is updated successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.updateRemoteLogSegmentMetadata(segmentMetadataUpdate).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.updateRemoteLogSegmentMetadata(segmentMetadataUpdate).get());
RemoteLogSegmentMetadata expectedSegMetadata = segmentMetadata.createWithUpdates(segmentMetadataUpdate);
// Check that the segment exists in RLMM.
- Optional<RemoteLogSegmentMetadata> segMetadataForOffset30Epoch1 = remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 1, 30L);
- Assertions.assertEquals(Optional.of(expectedSegMetadata), segMetadataForOffset30Epoch1);
+ Optional<RemoteLogSegmentMetadata> segMetadataForOffset30Epoch1 =
+ remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 1, 30L);
+ assertEquals(Optional.of(expectedSegMetadata), segMetadataForOffset30Epoch1);
// Mark the partition for deletion and wait for it to be updated successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.putRemotePartitionDeleteMetadata(
- createRemotePartitionDeleteMetadata(RemotePartitionDeleteState.DELETE_PARTITION_MARKED)).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.putRemotePartitionDeleteMetadata(
+ createRemotePartitionDeleteMetadata(DELETE_PARTITION_MARKED)).get());
- Optional<RemoteLogSegmentMetadata> segmentMetadataAfterDelMark = remoteLogMetadataManager.remoteLogSegmentMetadata(TP0,
- 1, 30L);
- Assertions.assertEquals(Optional.of(expectedSegMetadata), segmentMetadataAfterDelMark);
+ Optional<RemoteLogSegmentMetadata> segmentMetadataAfterDelMark =
+ remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 1, 30L);
+ assertEquals(Optional.of(expectedSegMetadata), segmentMetadataAfterDelMark);
// Set the partition deletion state as started. Partition and segments should still be accessible as they are not
// yet deleted. Wait until the segment state is updated successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.putRemotePartitionDeleteMetadata(
- createRemotePartitionDeleteMetadata(RemotePartitionDeleteState.DELETE_PARTITION_STARTED)).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.putRemotePartitionDeleteMetadata(
+ createRemotePartitionDeleteMetadata(DELETE_PARTITION_STARTED)).get());
- Optional<RemoteLogSegmentMetadata> segmentMetadataAfterDelStart = remoteLogMetadataManager.remoteLogSegmentMetadata(TP0,
- 1, 30L);
- Assertions.assertEquals(Optional.of(expectedSegMetadata), segmentMetadataAfterDelStart);
+ Optional<RemoteLogSegmentMetadata> segmentMetadataAfterDelStart =
+ remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 1, 30L);
+ assertEquals(Optional.of(expectedSegMetadata), segmentMetadataAfterDelStart);
// Set the partition deletion state as finished. RLMM should clear all its internal state for that partition.
// Wait until the segment state is updated successfully.
- Assertions.assertDoesNotThrow(() -> remoteLogMetadataManager.putRemotePartitionDeleteMetadata(
- createRemotePartitionDeleteMetadata(RemotePartitionDeleteState.DELETE_PARTITION_FINISHED)).get());
+ assertDoesNotThrow(() -> remoteLogMetadataManager.putRemotePartitionDeleteMetadata(
+ createRemotePartitionDeleteMetadata(DELETE_PARTITION_FINISHED)).get());
- Assertions.assertThrows(RemoteResourceNotFoundException.class,
+ assertThrows(RemoteResourceNotFoundException.class,
() -> remoteLogMetadataManager.remoteLogSegmentMetadata(TP0, 1, 30L));
}
}
(kafka) 10/12: KAFKA-16884 Refactor RemoteLogManagerConfig with AbstractConfig (#16199)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 9460e6b266a2f2fd3c4569ad3ad461877fe4f3ca
Author: Murali Basani <mu...@aiven.io>
AuthorDate: Thu Jun 6 18:06:25 2024 +0200
KAFKA-16884 Refactor RemoteLogManagerConfig with AbstractConfig (#16199)
Reviewers: Greg Harris <gh...@gmail.com>, Kamal Chandraprakash <ka...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
core/src/main/scala/kafka/server/KafkaConfig.scala | 4 +-
.../kafka/log/remote/RemoteLogManagerTest.java | 4 +-
.../unit/kafka/server/ReplicaManagerTest.scala | 8 +-
.../log/remote/storage/RemoteLogManagerConfig.java | 245 ++++-----------------
.../remote/storage/RemoteLogManagerConfigTest.java | 190 +++++-----------
5 files changed, 97 insertions(+), 354 deletions(-)
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 3635d5a3edd..06bce64f623 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -455,7 +455,7 @@ object KafkaConfig {
}
/** ********* Remote Log Management Configuration *********/
- RemoteLogManagerConfig.CONFIG_DEF.configKeys().values().forEach(key => configDef.define(key))
+ RemoteLogManagerConfig.configDef().configKeys().values().forEach(key => configDef.define(key))
def configNames: Seq[String] = configDef.names.asScala.toBuffer.sorted
private[server] def defaultValues: Map[String, _] = configDef.defaultValues.asScala
@@ -590,7 +590,7 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
val zkEnableSecureAcls: Boolean = getBoolean(ZkConfigs.ZK_ENABLE_SECURE_ACLS_CONFIG)
val zkMaxInFlightRequests: Int = getInt(ZkConfigs.ZK_MAX_IN_FLIGHT_REQUESTS_CONFIG)
- private val _remoteLogManagerConfig = new RemoteLogManagerConfig(this)
+ private val _remoteLogManagerConfig = new RemoteLogManagerConfig(props)
def remoteLogManagerConfig = _remoteLogManagerConfig
private def zkBooleanConfigOrSystemPropertyWithDefaultValue(propKey: String): Boolean = {
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 50b581fdf4e..9badaabf419 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -29,7 +29,6 @@ import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
import org.apache.kafka.common.compress.Compression;
-import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.errors.ReplicaNotAvailableException;
import org.apache.kafka.common.metrics.Metrics;
import org.apache.kafka.common.network.ListenerName;
@@ -2742,8 +2741,7 @@ public class RemoteLogManagerTest {
props.put(DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX + remoteLogMetadataConsumerTestProp, remoteLogMetadataConsumerTestVal);
props.put(DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX + remoteLogMetadataProducerTestProp, remoteLogMetadataProducerTestVal);
- AbstractConfig config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props);
- return new RemoteLogManagerConfig(config);
+ return new RemoteLogManagerConfig(props);
}
}
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 65cbcc7bd70..313b679c605 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -32,7 +32,7 @@ import kafka.zk.KafkaZkClient
import org.apache.kafka.clients.FetchSessionHandler
import org.apache.kafka.common.{DirectoryId, IsolationLevel, Node, TopicIdPartition, TopicPartition, Uuid}
import org.apache.kafka.common.compress.Compression
-import org.apache.kafka.common.config.{AbstractConfig, TopicConfig}
+import org.apache.kafka.common.config.{TopicConfig}
import org.apache.kafka.common.errors.{InvalidPidMappingException, KafkaStorageException}
import org.apache.kafka.common.message.LeaderAndIsrRequestData
import org.apache.kafka.common.message.LeaderAndIsrRequestData.LeaderAndIsrPartitionState
@@ -4092,8 +4092,7 @@ class ReplicaManagerTest {
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP, classOf[NoOpRemoteLogMetadataManager].getName)
// set log reader threads number to 2
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, 2.toString)
- val config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props)
- val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
+ val remoteLogManagerConfig = new RemoteLogManagerConfig(props)
val mockLog = mock(classOf[UnifiedLog])
val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem())
val remoteLogManager = new RemoteLogManager(
@@ -4192,8 +4191,7 @@ class ReplicaManagerTest {
props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, true.toString)
props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, classOf[NoOpRemoteStorageManager].getName)
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP, classOf[NoOpRemoteLogMetadataManager].getName)
- val config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props)
- val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
+ val remoteLogManagerConfig = new RemoteLogManagerConfig(props)
val dummyLog = mock(classOf[UnifiedLog])
val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem())
val remoteLogManager = new RemoteLogManager(
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index e41cc011c31..8224ab23825 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -20,9 +20,7 @@ import org.apache.kafka.common.config.AbstractConfig;
import org.apache.kafka.common.config.ConfigDef;
import java.util.Collections;
-import java.util.HashMap;
import java.util.Map;
-import java.util.Objects;
import static org.apache.kafka.common.config.ConfigDef.Importance.LOW;
import static org.apache.kafka.common.config.ConfigDef.Importance.MEDIUM;
@@ -34,7 +32,7 @@ import static org.apache.kafka.common.config.ConfigDef.Type.INT;
import static org.apache.kafka.common.config.ConfigDef.Type.LONG;
import static org.apache.kafka.common.config.ConfigDef.Type.STRING;
-public final class RemoteLogManagerConfig {
+public final class RemoteLogManagerConfig extends AbstractConfig {
/**
* Prefix used for properties to be passed to {@link RemoteStorageManager} implementation. Remote log subsystem collects all the properties having
@@ -188,10 +186,8 @@ public final class RemoteLogManagerConfig {
public static final String REMOTE_FETCH_MAX_WAIT_MS_DOC = "The maximum amount of time the server will wait before answering the remote fetch request";
public static final int DEFAULT_REMOTE_FETCH_MAX_WAIT_MS = 500;
- public static final ConfigDef CONFIG_DEF = new ConfigDef();
-
- static {
- CONFIG_DEF
+ public static ConfigDef configDef() {
+ return new ConfigDef()
.define(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP,
BOOLEAN,
DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE,
@@ -358,291 +354,128 @@ public final class RemoteLogManagerConfig {
REMOTE_FETCH_MAX_WAIT_MS_DOC);
}
- private final boolean enableRemoteStorageSystem;
- private final String remoteStorageManagerClassName;
- private final String remoteStorageManagerClassPath;
- private final String remoteLogMetadataManagerClassName;
- private final String remoteLogMetadataManagerClassPath;
- private final long remoteLogIndexFileCacheTotalSizeBytes;
- private final int remoteLogManagerThreadPoolSize;
- private final int remoteLogManagerCopierThreadPoolSize;
- private final int remoteLogManagerExpirationThreadPoolSize;
- private final long remoteLogManagerTaskIntervalMs;
- private final long remoteLogManagerTaskRetryBackoffMs;
- private final long remoteLogManagerTaskRetryBackoffMaxMs;
- private final double remoteLogManagerTaskRetryJitter;
- private final int remoteLogReaderThreads;
- private final int remoteLogReaderMaxPendingTasks;
- private final String remoteStorageManagerPrefix;
- private final HashMap<String, Object> remoteStorageManagerProps;
- private final String remoteLogMetadataManagerPrefix;
- private final HashMap<String, Object> remoteLogMetadataManagerProps;
- private final String remoteLogMetadataManagerListenerName;
- private final int remoteLogMetadataCustomMetadataMaxBytes;
- private final long remoteLogManagerCopyMaxBytesPerSecond;
- private final int remoteLogManagerCopyNumQuotaSamples;
- private final int remoteLogManagerCopyQuotaWindowSizeSeconds;
- private final long remoteLogManagerFetchMaxBytesPerSecond;
- private final int remoteLogManagerFetchNumQuotaSamples;
- private final int remoteLogManagerFetchQuotaWindowSizeSeconds;
- private final int remoteFetchMaxWaitMs;
-
- public RemoteLogManagerConfig(AbstractConfig config) {
- this(config.getBoolean(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP),
- config.getString(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP),
- config.getString(REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP),
- config.getString(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP),
- config.getString(REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP),
- config.getString(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP),
- config.getLong(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP),
- config.getInt(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP),
- config.getInt(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP),
- config.getInt(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP),
- config.getLong(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP),
- config.getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP),
- config.getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP),
- config.getDouble(REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP),
- config.getInt(REMOTE_LOG_READER_THREADS_PROP),
- config.getInt(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP),
- config.getInt(REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP),
- config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP),
- config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP) != null
- ? config.originalsWithPrefix(config.getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP))
- : Collections.emptyMap(),
- config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP),
- config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP) != null
- ? config.originalsWithPrefix(config.getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP))
- : Collections.emptyMap(),
- config.getLong(REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP),
- config.getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_PROP),
- config.getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_PROP),
- config.getLong(REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP),
- config.getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP),
- config.getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP),
- config.getInt(REMOTE_FETCH_MAX_WAIT_MS_PROP));
- }
-
- // Visible for testing
- public RemoteLogManagerConfig(boolean enableRemoteStorageSystem,
- String remoteStorageManagerClassName,
- String remoteStorageManagerClassPath,
- String remoteLogMetadataManagerClassName,
- String remoteLogMetadataManagerClassPath,
- String remoteLogMetadataManagerListenerName,
- long remoteLogIndexFileCacheTotalSizeBytes,
- int remoteLogManagerThreadPoolSize,
- int remoteLogManagerCopierThreadPoolSize,
- int remoteLogManagerExpirationThreadPoolSize,
- long remoteLogManagerTaskIntervalMs,
- long remoteLogManagerTaskRetryBackoffMs,
- long remoteLogManagerTaskRetryBackoffMaxMs,
- double remoteLogManagerTaskRetryJitter,
- int remoteLogReaderThreads,
- int remoteLogReaderMaxPendingTasks,
- int remoteLogMetadataCustomMetadataMaxBytes,
- String remoteStorageManagerPrefix,
- Map<String, Object> remoteStorageManagerProps, /* properties having keys stripped out with remoteStorageManagerPrefix */
- String remoteLogMetadataManagerPrefix,
- Map<String, Object> remoteLogMetadataManagerProps, /* properties having keys stripped out with remoteLogMetadataManagerPrefix */
- long remoteLogManagerCopyMaxBytesPerSecond,
- int remoteLogManagerCopyNumQuotaSamples,
- int remoteLogManagerCopyQuotaWindowSizeSeconds,
- long remoteLogManagerFetchMaxBytesPerSecond,
- int remoteLogManagerFetchNumQuotaSamples,
- int remoteLogManagerFetchQuotaWindowSizeSeconds,
- int remoteFetchMaxWaitMs) {
- this.enableRemoteStorageSystem = enableRemoteStorageSystem;
- this.remoteStorageManagerClassName = remoteStorageManagerClassName;
- this.remoteStorageManagerClassPath = remoteStorageManagerClassPath;
- this.remoteLogMetadataManagerClassName = remoteLogMetadataManagerClassName;
- this.remoteLogMetadataManagerClassPath = remoteLogMetadataManagerClassPath;
- this.remoteLogIndexFileCacheTotalSizeBytes = remoteLogIndexFileCacheTotalSizeBytes;
- this.remoteLogManagerThreadPoolSize = remoteLogManagerThreadPoolSize;
- this.remoteLogManagerCopierThreadPoolSize = remoteLogManagerCopierThreadPoolSize;
- this.remoteLogManagerExpirationThreadPoolSize = remoteLogManagerExpirationThreadPoolSize;
- this.remoteLogManagerTaskIntervalMs = remoteLogManagerTaskIntervalMs;
- this.remoteLogManagerTaskRetryBackoffMs = remoteLogManagerTaskRetryBackoffMs;
- this.remoteLogManagerTaskRetryBackoffMaxMs = remoteLogManagerTaskRetryBackoffMaxMs;
- this.remoteLogManagerTaskRetryJitter = remoteLogManagerTaskRetryJitter;
- this.remoteLogReaderThreads = remoteLogReaderThreads;
- this.remoteLogReaderMaxPendingTasks = remoteLogReaderMaxPendingTasks;
- this.remoteStorageManagerPrefix = remoteStorageManagerPrefix;
- this.remoteStorageManagerProps = new HashMap<>(remoteStorageManagerProps);
- this.remoteLogMetadataManagerPrefix = remoteLogMetadataManagerPrefix;
- this.remoteLogMetadataManagerProps = new HashMap<>(remoteLogMetadataManagerProps);
- this.remoteLogMetadataManagerListenerName = remoteLogMetadataManagerListenerName;
- this.remoteLogMetadataCustomMetadataMaxBytes = remoteLogMetadataCustomMetadataMaxBytes;
- this.remoteLogManagerCopyMaxBytesPerSecond = remoteLogManagerCopyMaxBytesPerSecond;
- this.remoteLogManagerCopyNumQuotaSamples = remoteLogManagerCopyNumQuotaSamples;
- this.remoteLogManagerCopyQuotaWindowSizeSeconds = remoteLogManagerCopyQuotaWindowSizeSeconds;
- this.remoteLogManagerFetchMaxBytesPerSecond = remoteLogManagerFetchMaxBytesPerSecond;
- this.remoteLogManagerFetchNumQuotaSamples = remoteLogManagerFetchNumQuotaSamples;
- this.remoteLogManagerFetchQuotaWindowSizeSeconds = remoteLogManagerFetchQuotaWindowSizeSeconds;
- this.remoteFetchMaxWaitMs = remoteFetchMaxWaitMs;
+ public RemoteLogManagerConfig(Map<?, ?> props) {
+ super(configDef(), props);
}
public boolean enableRemoteStorageSystem() {
- return enableRemoteStorageSystem;
+ return getBoolean(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP);
}
public String remoteStorageManagerClassName() {
- return remoteStorageManagerClassName;
+ return getString(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP);
}
public String remoteStorageManagerClassPath() {
- return remoteStorageManagerClassPath;
+ return getString(REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP);
}
public String remoteLogMetadataManagerClassName() {
- return remoteLogMetadataManagerClassName;
+ return getString(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP);
}
public String remoteLogMetadataManagerClassPath() {
- return remoteLogMetadataManagerClassPath;
+ return getString(REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP);
}
public long remoteLogIndexFileCacheTotalSizeBytes() {
- return remoteLogIndexFileCacheTotalSizeBytes;
+ return getLong(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP);
}
public int remoteLogManagerThreadPoolSize() {
- return remoteLogManagerThreadPoolSize;
+ return getInt(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP);
}
public int remoteLogManagerCopierThreadPoolSize() {
- return remoteLogManagerCopierThreadPoolSize;
+ return getInt(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP);
}
public int remoteLogManagerExpirationThreadPoolSize() {
- return remoteLogManagerExpirationThreadPoolSize;
+ return getInt(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP);
}
public long remoteLogManagerTaskIntervalMs() {
- return remoteLogManagerTaskIntervalMs;
+ return getLong(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP);
}
public long remoteLogManagerTaskRetryBackoffMs() {
- return remoteLogManagerTaskRetryBackoffMs;
+ return getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP);
}
public long remoteLogManagerTaskRetryBackoffMaxMs() {
- return remoteLogManagerTaskRetryBackoffMaxMs;
+ return getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP);
}
public double remoteLogManagerTaskRetryJitter() {
- return remoteLogManagerTaskRetryJitter;
+ return getDouble(REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP);
}
public int remoteLogReaderThreads() {
- return remoteLogReaderThreads;
+ return getInt(REMOTE_LOG_READER_THREADS_PROP);
}
public int remoteLogReaderMaxPendingTasks() {
- return remoteLogReaderMaxPendingTasks;
+ return getInt(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP);
}
public String remoteLogMetadataManagerListenerName() {
- return remoteLogMetadataManagerListenerName;
+ return getString(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP);
}
public int remoteLogMetadataCustomMetadataMaxBytes() {
- return remoteLogMetadataCustomMetadataMaxBytes;
+ return getInt(REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP);
}
public String remoteStorageManagerPrefix() {
- return remoteStorageManagerPrefix;
+ return getString(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP);
}
public String remoteLogMetadataManagerPrefix() {
- return remoteLogMetadataManagerPrefix;
+ return getString(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP);
}
public Map<String, Object> remoteStorageManagerProps() {
- return Collections.unmodifiableMap(remoteStorageManagerProps);
+ return getConfigProps(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP);
}
public Map<String, Object> remoteLogMetadataManagerProps() {
- return Collections.unmodifiableMap(remoteLogMetadataManagerProps);
+ return getConfigProps(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP);
+ }
+
+ public Map<String, Object> getConfigProps(String configPrefixProp) {
+ String prefixProp = getString(configPrefixProp);
+ return prefixProp == null ? Collections.emptyMap() : Collections.unmodifiableMap(originalsWithPrefix(prefixProp));
}
public long remoteLogManagerCopyMaxBytesPerSecond() {
- return remoteLogManagerCopyMaxBytesPerSecond;
+ return getLong(REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP);
}
public int remoteLogManagerCopyNumQuotaSamples() {
- return remoteLogManagerCopyNumQuotaSamples;
+ return getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_PROP);
}
public int remoteLogManagerCopyQuotaWindowSizeSeconds() {
- return remoteLogManagerCopyQuotaWindowSizeSeconds;
+ return getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_PROP);
}
public long remoteLogManagerFetchMaxBytesPerSecond() {
- return remoteLogManagerFetchMaxBytesPerSecond;
+ return getLong(REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP);
}
public int remoteLogManagerFetchNumQuotaSamples() {
- return remoteLogManagerFetchNumQuotaSamples;
+ return getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP);
}
public int remoteLogManagerFetchQuotaWindowSizeSeconds() {
- return remoteLogManagerFetchQuotaWindowSizeSeconds;
+ return getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP);
}
public int remoteFetchMaxWaitMs() {
- return remoteFetchMaxWaitMs;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) return true;
- if (!(o instanceof RemoteLogManagerConfig)) return false;
- RemoteLogManagerConfig that = (RemoteLogManagerConfig) o;
- return enableRemoteStorageSystem == that.enableRemoteStorageSystem
- && remoteLogIndexFileCacheTotalSizeBytes == that.remoteLogIndexFileCacheTotalSizeBytes
- && remoteLogManagerThreadPoolSize == that.remoteLogManagerThreadPoolSize
- && remoteLogManagerCopierThreadPoolSize == that.remoteLogManagerCopierThreadPoolSize
- && remoteLogManagerExpirationThreadPoolSize == that.remoteLogManagerExpirationThreadPoolSize
- && remoteLogManagerTaskIntervalMs == that.remoteLogManagerTaskIntervalMs
- && remoteLogManagerTaskRetryBackoffMs == that.remoteLogManagerTaskRetryBackoffMs
- && remoteLogManagerTaskRetryBackoffMaxMs == that.remoteLogManagerTaskRetryBackoffMaxMs
- && remoteLogManagerTaskRetryJitter == that.remoteLogManagerTaskRetryJitter
- && remoteLogReaderThreads == that.remoteLogReaderThreads
- && remoteLogReaderMaxPendingTasks == that.remoteLogReaderMaxPendingTasks
- && remoteLogMetadataCustomMetadataMaxBytes == that.remoteLogMetadataCustomMetadataMaxBytes
- && Objects.equals(remoteStorageManagerClassName, that.remoteStorageManagerClassName)
- && Objects.equals(remoteStorageManagerClassPath, that.remoteStorageManagerClassPath)
- && Objects.equals(remoteLogMetadataManagerClassName, that.remoteLogMetadataManagerClassName)
- && Objects.equals(remoteLogMetadataManagerClassPath, that.remoteLogMetadataManagerClassPath)
- && Objects.equals(remoteLogMetadataManagerListenerName, that.remoteLogMetadataManagerListenerName)
- && Objects.equals(remoteStorageManagerProps, that.remoteStorageManagerProps)
- && Objects.equals(remoteLogMetadataManagerProps, that.remoteLogMetadataManagerProps)
- && Objects.equals(remoteStorageManagerPrefix, that.remoteStorageManagerPrefix)
- && Objects.equals(remoteLogMetadataManagerPrefix, that.remoteLogMetadataManagerPrefix)
- && remoteLogManagerCopyMaxBytesPerSecond == that.remoteLogManagerCopyMaxBytesPerSecond
- && remoteLogManagerCopyNumQuotaSamples == that.remoteLogManagerCopyNumQuotaSamples
- && remoteLogManagerCopyQuotaWindowSizeSeconds == that.remoteLogManagerCopyQuotaWindowSizeSeconds
- && remoteLogManagerFetchMaxBytesPerSecond == that.remoteLogManagerFetchMaxBytesPerSecond
- && remoteLogManagerFetchNumQuotaSamples == that.remoteLogManagerFetchNumQuotaSamples
- && remoteLogManagerFetchQuotaWindowSizeSeconds == that.remoteLogManagerFetchQuotaWindowSizeSeconds
- && remoteFetchMaxWaitMs == that.remoteFetchMaxWaitMs;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(
- enableRemoteStorageSystem, remoteStorageManagerClassName, remoteStorageManagerClassPath,
- remoteLogMetadataManagerClassName, remoteLogMetadataManagerClassPath, remoteLogMetadataManagerListenerName,
- remoteLogMetadataCustomMetadataMaxBytes, remoteLogIndexFileCacheTotalSizeBytes, remoteLogManagerThreadPoolSize,
- remoteLogManagerCopierThreadPoolSize, remoteLogManagerExpirationThreadPoolSize, remoteLogManagerTaskIntervalMs,
- remoteLogManagerTaskRetryBackoffMs, remoteLogManagerTaskRetryBackoffMaxMs, remoteLogManagerTaskRetryJitter,
- remoteLogReaderThreads, remoteLogReaderMaxPendingTasks, remoteStorageManagerProps, remoteLogMetadataManagerProps,
- remoteStorageManagerPrefix, remoteLogMetadataManagerPrefix, remoteLogManagerCopyMaxBytesPerSecond,
- remoteLogManagerCopyNumQuotaSamples, remoteLogManagerCopyQuotaWindowSizeSeconds, remoteLogManagerFetchMaxBytesPerSecond,
- remoteLogManagerFetchNumQuotaSamples, remoteLogManagerFetchQuotaWindowSizeSeconds, remoteFetchMaxWaitMs);
+ return getInt(REMOTE_FETCH_MAX_WAIT_MS_PROP);
}
public static void main(String[] args) {
- System.out.println(CONFIG_DEF.toHtml(4, config -> "remote_log_manager_" + config));
+ System.out.println(configDef().toHtml(4, config -> "remote_log_manager_" + config));
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
index 058f81626d1..3e6702f84c0 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
@@ -16,184 +16,98 @@
*/
package org.apache.kafka.server.log.remote.storage;
-import org.apache.kafka.common.config.AbstractConfig;
+import org.apache.kafka.common.config.ConfigException;
import org.junit.jupiter.api.Test;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.ValueSource;
import java.util.Collections;
import java.util.HashMap;
import java.util.Map;
-import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME;
-import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP;
import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertNotEquals;
+import static org.junit.jupiter.api.Assertions.assertThrows;
public class RemoteLogManagerConfigTest {
- private static class TestConfig extends AbstractConfig {
- public TestConfig(Map<?, ?> originals) {
- super(RemoteLogManagerConfig.CONFIG_DEF, originals, true);
- }
- }
-
- @ParameterizedTest
- @ValueSource(booleans = {true, false})
- public void testValidConfigs(boolean useDefaultRemoteLogMetadataManagerClass) {
+ @Test
+ public void testValidConfigs() {
String rsmPrefix = "__custom.rsm.";
String rlmmPrefix = "__custom.rlmm.";
Map<String, Object> rsmProps = Collections.singletonMap("rsm.prop", "val");
Map<String, Object> rlmmProps = Collections.singletonMap("rlmm.prop", "val");
- RemoteLogManagerConfig expectedRemoteLogManagerConfig = getRemoteLogManagerConfig(useDefaultRemoteLogMetadataManagerClass,
- rsmPrefix,
- rlmmPrefix,
- rsmProps,
- rlmmProps);
- Map<String, Object> props = extractProps(expectedRemoteLogManagerConfig);
+ Map<String, Object> props = getRLMProps(rsmPrefix, rlmmPrefix);
rsmProps.forEach((k, v) -> props.put(rsmPrefix + k, v));
rlmmProps.forEach((k, v) -> props.put(rlmmPrefix + k, v));
+
+ RemoteLogManagerConfig expectedRemoteLogManagerConfig = new RemoteLogManagerConfig(props);
+
// Removing remote.log.metadata.manager.class.name so that the default value gets picked up.
- if (useDefaultRemoteLogMetadataManagerClass) {
- props.remove(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP);
- }
- TestConfig config = new TestConfig(props);
- RemoteLogManagerConfig remoteLogManagerConfig = new RemoteLogManagerConfig(config);
- assertEquals(expectedRemoteLogManagerConfig, remoteLogManagerConfig);
+ props.remove(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP);
+
+ RemoteLogManagerConfig remoteLogManagerConfig = new RemoteLogManagerConfig(props);
+ assertEquals(expectedRemoteLogManagerConfig.values(), remoteLogManagerConfig.values());
+
+ assertEquals(rsmProps, remoteLogManagerConfig.remoteStorageManagerProps());
+ assertEquals(rlmmProps, remoteLogManagerConfig.remoteLogMetadataManagerProps());
}
- private static RemoteLogManagerConfig getRemoteLogManagerConfig(boolean useDefaultRemoteLogMetadataManagerClass,
- String rsmPrefix,
- String rlmmPrefix,
- Map<String, Object> rsmProps,
- Map<String, Object> rlmmProps) {
- String remoteLogMetadataManagerClass = useDefaultRemoteLogMetadataManagerClass ? DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME : "dummy.remote.log.metadata.class";
- return new RemoteLogManagerConfig(true,
- "dummy.remote.storage.class",
- "dummy.remote.storage.class.path",
- remoteLogMetadataManagerClass,
- "dummy.remote.log.metadata.class.path",
- "listener.name",
- 1024 * 1024L,
- 1,
- 1,
- 1,
- 60000L,
- 100L,
- 60000L,
- 0.3,
- 10,
- 100,
- 100,
- rsmPrefix,
- rsmProps,
- rlmmPrefix,
- rlmmProps,
- Long.MAX_VALUE,
- 11,
- 1,
- Long.MAX_VALUE,
- 11,
- 1,
- 500);
+ @Test
+ public void testDefaultConfigs() {
+ // Even with empty properties, RemoteLogManagerConfig has default values
+ Map<String, Object> emptyProps = new HashMap<>();
+ RemoteLogManagerConfig remoteLogManagerConfigEmptyConfig = new RemoteLogManagerConfig(emptyProps);
+ assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE, remoteLogManagerConfigEmptyConfig.remoteLogManagerThreadPoolSize());
+ assertEquals(RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM, remoteLogManagerConfigEmptyConfig.remoteLogManagerCopyNumQuotaSamples());
+ }
+
+ @Test
+ public void testValidateEmptyStringConfig() {
+ // Test with a empty string props should throw ConfigException
+ Map<String, Object> emptyStringProps = Collections.singletonMap(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, "");
+ assertThrows(ConfigException.class, () ->
+ new RemoteLogManagerConfig(emptyStringProps));
}
- private Map<String, Object> extractProps(RemoteLogManagerConfig remoteLogManagerConfig) {
+ private Map<String, Object> getRLMProps(String rsmPrefix, String rlmmPrefix) {
+
Map<String, Object> props = new HashMap<>();
- props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP,
- remoteLogManagerConfig.enableRemoteStorageSystem());
+ props.put(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, true);
props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP,
- remoteLogManagerConfig.remoteStorageManagerClassName());
+ "dummy.remote.storage.class");
props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP,
- remoteLogManagerConfig.remoteStorageManagerClassPath());
+ "dummy.remote.storage.class.path");
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
- remoteLogManagerConfig.remoteLogMetadataManagerClassName());
+ RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME);
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP,
- remoteLogManagerConfig.remoteLogMetadataManagerClassPath());
+ "dummy.remote.log.metadata.class.path");
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP,
- remoteLogManagerConfig.remoteLogMetadataManagerListenerName());
+ "listener.name");
props.put(RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP,
- remoteLogManagerConfig.remoteLogIndexFileCacheTotalSizeBytes());
+ 1024 * 1024L);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP,
- remoteLogManagerConfig.remoteLogManagerThreadPoolSize());
+ 1);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
- remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize());
+ 1);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
- remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize());
+ 1);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP,
- remoteLogManagerConfig.remoteLogManagerTaskIntervalMs());
+ 60000L);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP,
- remoteLogManagerConfig.remoteLogManagerTaskRetryBackoffMs());
+ 100L);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP,
- remoteLogManagerConfig.remoteLogManagerTaskRetryBackoffMaxMs());
+ 60000L);
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP,
- remoteLogManagerConfig.remoteLogManagerTaskRetryJitter());
+ 0.3);
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP,
- remoteLogManagerConfig.remoteLogReaderThreads());
+ 10);
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP,
- remoteLogManagerConfig.remoteLogReaderMaxPendingTasks());
+ 100);
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP,
- remoteLogManagerConfig.remoteLogMetadataCustomMetadataMaxBytes());
+ 100);
props.put(RemoteLogManagerConfig.REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP,
- remoteLogManagerConfig.remoteStorageManagerPrefix());
+ rsmPrefix);
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP,
- remoteLogManagerConfig.remoteLogMetadataManagerPrefix());
+ rlmmPrefix);
return props;
}
-
- @Test
- public void testHashCodeAndEquals_ForAllAndTwoFields() {
- String rsmPrefix = "__custom.rsm.";
- String rlmmPrefix = "__custom.rlmm.";
- Map<String, Object> rsmProps = Collections.singletonMap("rsm.prop", "val");
- Map<String, Object> rlmmProps = Collections.singletonMap("rlmm.prop", "val");
- RemoteLogManagerConfig config1 = getRemoteLogManagerConfig(false,
- rsmPrefix,
- rlmmPrefix,
- rsmProps,
- rlmmProps);
- RemoteLogManagerConfig config2 = getRemoteLogManagerConfig(false,
- rsmPrefix,
- rlmmPrefix,
- rsmProps,
- rlmmProps);
-
- // Initially, hash codes should be equal for default objects
- assertEquals(config1.hashCode(), config2.hashCode());
-
- // Initially, objects should be equal
- assertEquals(config1, config2);
-
- // Test for specific field remoteLogManagerCopierThreadPoolSize
- RemoteLogManagerConfig config3 = new RemoteLogManagerConfig(true, "dummy.remote.storage.class",
- "dummy.remote.storage.class.path",
- "dummy.remote.log.metadata.class", "dummy.remote.log.metadata.class.path",
- "listener.name",
- 1024 * 1024L,
- 1,
- 2, // Change here
- 2, // Change here
- 60000L,
- 100L,
- 60000L,
- 0.3,
- 10,
- 100,
- 100,
- rsmPrefix,
- rsmProps,
- rlmmPrefix,
- rlmmProps,
- Long.MAX_VALUE,
- 11,
- 1,
- Long.MAX_VALUE,
- 11,
- 1,
- 500);
-
- assertNotEquals(config1.hashCode(), config3.hashCode());
- assertNotEquals(config1, config3);
- }
-}
+}
\ No newline at end of file
(kafka) 08/12: KAFKA-15776: Introduce remote.fetch.max.timeout.ms to configure DelayedRemoteFetch timeout (#14778)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit b6848d699da3299b55273b707124f2dbfa552743
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Wed Jun 5 12:12:23 2024 +0530
KAFKA-15776: Introduce remote.fetch.max.timeout.ms to configure DelayedRemoteFetch timeout (#14778)
KIP-1018, part1, Introduce remote.fetch.max.timeout.ms to configure DelayedRemoteFetch timeout
Reviewers: Luke Chen <sh...@gmail.com>
---
.../kafka/clients/consumer/ConsumerConfig.java | 5 +-
.../scala/kafka/server/DelayedRemoteFetch.scala | 3 +-
.../main/scala/kafka/server/ReplicaManager.scala | 3 +-
.../kafka/server/DelayedRemoteFetchTest.scala | 21 +-
.../log/remote/storage/RemoteLogManagerConfig.java | 342 +++++++++++----------
.../remote/storage/RemoteLogManagerConfigTest.java | 6 +-
6 files changed, 203 insertions(+), 177 deletions(-)
diff --git a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
index 7ec147e9e3c..76bfe7e91a1 100644
--- a/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
+++ b/clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
@@ -198,7 +198,10 @@ public class ConsumerConfig extends AbstractConfig {
* <code>fetch.max.wait.ms</code>
*/
public static final String FETCH_MAX_WAIT_MS_CONFIG = "fetch.max.wait.ms";
- private static final String FETCH_MAX_WAIT_MS_DOC = "The maximum amount of time the server will block before answering the fetch request if there isn't sufficient data to immediately satisfy the requirement given by fetch.min.bytes.";
+ private static final String FETCH_MAX_WAIT_MS_DOC = "The maximum amount of time the server will block before " +
+ "answering the fetch request there isn't sufficient data to immediately satisfy the requirement given by " +
+ "fetch.min.bytes. This config is used only for local log fetch. To tune the remote fetch maximum wait " +
+ "time, please refer to 'remote.fetch.max.wait.ms' broker config";
public static final int DEFAULT_FETCH_MAX_WAIT_MS = 500;
/** <code>metadata.max.age.ms</code> */
diff --git a/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala b/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala
index 00d6afb89ff..58a866aa4a6 100644
--- a/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala
+++ b/core/src/main/scala/kafka/server/DelayedRemoteFetch.scala
@@ -35,12 +35,13 @@ import scala.collection._
class DelayedRemoteFetch(remoteFetchTask: Future[Void],
remoteFetchResult: CompletableFuture[RemoteLogReadResult],
remoteFetchInfo: RemoteStorageFetchInfo,
+ remoteFetchMaxWaitMs: Long,
fetchPartitionStatus: Seq[(TopicIdPartition, FetchPartitionStatus)],
fetchParams: FetchParams,
localReadResults: Seq[(TopicIdPartition, LogReadResult)],
replicaManager: ReplicaManager,
responseCallback: Seq[(TopicIdPartition, FetchPartitionData)] => Unit)
- extends DelayedOperation(fetchParams.maxWaitMs) {
+ extends DelayedOperation(remoteFetchMaxWaitMs) {
if (fetchParams.isFromFollower) {
throw new IllegalStateException(s"The follower should not invoke remote fetch. Fetch params are: $fetchParams")
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index 6908ec096d1..17e9a5c1b9e 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -1479,7 +1479,8 @@ class ReplicaManager(val config: KafkaConfig,
return Some(createLogReadResult(e))
}
- val remoteFetch = new DelayedRemoteFetch(remoteFetchTask, remoteFetchResult, remoteFetchInfo,
+ val remoteFetchMaxWaitMs = config.remoteLogManagerConfig.remoteFetchMaxWaitMs()
+ val remoteFetch = new DelayedRemoteFetch(remoteFetchTask, remoteFetchResult, remoteFetchInfo, remoteFetchMaxWaitMs,
fetchPartitionStatus, params, logReadResults, this, responseCallback)
delayedRemoteFetchPurgatory.tryCompleteElseWatch(remoteFetch, Seq(key))
diff --git a/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala b/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala
index c35385bcbc8..ea1ffaf0b11 100644
--- a/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala
+++ b/core/src/test/scala/integration/kafka/server/DelayedRemoteFetchTest.scala
@@ -40,6 +40,7 @@ class DelayedRemoteFetchTest {
private val fetchOffset = 500L
private val logStartOffset = 0L
private val currentLeaderEpoch = Optional.of[Integer](10)
+ private val remoteFetchMaxWaitMs = 500
private val fetchStatus = FetchPartitionStatus(
startOffsetMetadata = new LogOffsetMetadata(fetchOffset),
@@ -64,8 +65,8 @@ class DelayedRemoteFetchTest {
val leaderLogStartOffset = 10
val logReadInfo = buildReadResult(Errors.NONE, highWatermark, leaderLogStartOffset)
- val delayedRemoteFetch = new DelayedRemoteFetch(null, future, fetchInfo, Seq(topicIdPartition -> fetchStatus), fetchParams,
- Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
+ val delayedRemoteFetch = new DelayedRemoteFetch(null, future, fetchInfo, remoteFetchMaxWaitMs,
+ Seq(topicIdPartition -> fetchStatus), fetchParams, Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
when(replicaManager.getPartitionOrException(topicIdPartition.topicPartition))
.thenReturn(mock(classOf[Partition]))
@@ -100,8 +101,8 @@ class DelayedRemoteFetchTest {
val leaderLogStartOffset = 10
val logReadInfo = buildReadResult(Errors.NONE, highWatermark, leaderLogStartOffset)
val fetchParams = buildFetchParams(replicaId = 1, maxWaitMs = 500)
- assertThrows(classOf[IllegalStateException], () => new DelayedRemoteFetch(null, future, fetchInfo, Seq(topicIdPartition -> fetchStatus), fetchParams,
- Seq(topicIdPartition -> logReadInfo), replicaManager, callback))
+ assertThrows(classOf[IllegalStateException], () => new DelayedRemoteFetch(null, future, fetchInfo, remoteFetchMaxWaitMs,
+ Seq(topicIdPartition -> fetchStatus), fetchParams, Seq(topicIdPartition -> logReadInfo), replicaManager, callback))
}
@Test
@@ -124,8 +125,8 @@ class DelayedRemoteFetchTest {
val logReadInfo = buildReadResult(Errors.NONE)
- val delayedRemoteFetch = new DelayedRemoteFetch(null, future, fetchInfo, Seq(topicIdPartition -> fetchStatus), fetchParams,
- Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
+ val delayedRemoteFetch = new DelayedRemoteFetch(null, future, fetchInfo, remoteFetchMaxWaitMs,
+ Seq(topicIdPartition -> fetchStatus), fetchParams, Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
// delayed remote fetch should still be able to complete
assertTrue(delayedRemoteFetch.tryComplete())
@@ -155,8 +156,8 @@ class DelayedRemoteFetchTest {
// build a read result with error
val logReadInfo = buildReadResult(Errors.FENCED_LEADER_EPOCH)
- val delayedRemoteFetch = new DelayedRemoteFetch(null, future, fetchInfo, Seq(topicIdPartition -> fetchStatus), fetchParams,
- Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
+ val delayedRemoteFetch = new DelayedRemoteFetch(null, future, fetchInfo, remoteFetchMaxWaitMs,
+ Seq(topicIdPartition -> fetchStatus), fetchParams, Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
assertTrue(delayedRemoteFetch.tryComplete())
assertTrue(delayedRemoteFetch.isCompleted)
@@ -184,8 +185,8 @@ class DelayedRemoteFetchTest {
val fetchInfo: RemoteStorageFetchInfo = new RemoteStorageFetchInfo(0, false, topicIdPartition.topicPartition(), null, null, false)
val logReadInfo = buildReadResult(Errors.NONE, highWatermark, leaderLogStartOffset)
- val delayedRemoteFetch = new DelayedRemoteFetch(remoteFetchTask, future, fetchInfo, Seq(topicIdPartition -> fetchStatus), fetchParams,
- Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
+ val delayedRemoteFetch = new DelayedRemoteFetch(remoteFetchTask, future, fetchInfo, remoteFetchMaxWaitMs,
+ Seq(topicIdPartition -> fetchStatus), fetchParams, Seq(topicIdPartition -> logReadInfo), replicaManager, callback)
when(replicaManager.getPartitionOrException(topicIdPartition.topicPartition))
.thenReturn(mock(classOf[Partition]))
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index 1add933d788..e41cc011c31 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -54,7 +54,6 @@ public final class RemoteLogManagerConfig {
"implementation. For example this value can be `rlmm.config.`.";
public static final String DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX = "rlmm.config.";
-
public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP = "remote.log.storage.system.enable";
public static final String REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC = "Whether to enable tiered storage functionality in a broker or not. Valid values " +
"are `true` or `false` and the default value is false. When it is true broker starts all the services required for the tiered storage functionality.";
@@ -185,167 +184,178 @@ public final class RemoteLogManagerConfig {
"The default value is 1 second.";
public static final int DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS = 1;
+ public static final String REMOTE_FETCH_MAX_WAIT_MS_PROP = "remote.fetch.max.wait.ms";
+ public static final String REMOTE_FETCH_MAX_WAIT_MS_DOC = "The maximum amount of time the server will wait before answering the remote fetch request";
+ public static final int DEFAULT_REMOTE_FETCH_MAX_WAIT_MS = 500;
+
public static final ConfigDef CONFIG_DEF = new ConfigDef();
static {
- CONFIG_DEF.define(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP,
- BOOLEAN,
- DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE,
- null,
- MEDIUM,
- REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC)
- .define(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP,
- STRING,
- DEFAULT_REMOTE_STORAGE_MANAGER_CONFIG_PREFIX,
- new ConfigDef.NonEmptyString(),
- MEDIUM,
- REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC)
- .define(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP,
- STRING,
- DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX,
- new ConfigDef.NonEmptyString(),
- MEDIUM,
- REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_DOC)
- .define(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, STRING,
- null,
- new ConfigDef.NonEmptyString(),
- MEDIUM,
- REMOTE_STORAGE_MANAGER_CLASS_NAME_DOC)
- .define(REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP, STRING,
- null,
- null,
- MEDIUM,
- REMOTE_STORAGE_MANAGER_CLASS_PATH_DOC)
- .define(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
- STRING,
- DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME,
- new ConfigDef.NonEmptyString(),
- MEDIUM,
- REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_DOC)
- .define(REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP,
- STRING,
- null,
- null,
- MEDIUM,
- REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_DOC)
- .define(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, STRING,
- null,
- new ConfigDef.NonEmptyString(),
- MEDIUM,
- REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC)
- .define(REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP,
- INT,
- DEFAULT_REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES,
- atLeast(0),
- LOW,
- REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_DOC)
- .define(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP,
- LONG,
- DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES,
- atLeast(1),
- LOW,
- REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC)
- .define(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP,
- INT,
- DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC)
- .defineInternal(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ CONFIG_DEF
+ .define(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP,
+ BOOLEAN,
+ DEFAULT_REMOTE_LOG_STORAGE_SYSTEM_ENABLE,
+ null,
+ MEDIUM,
+ REMOTE_LOG_STORAGE_SYSTEM_ENABLE_DOC)
+ .define(REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_PROP,
+ STRING,
+ DEFAULT_REMOTE_STORAGE_MANAGER_CONFIG_PREFIX,
+ new ConfigDef.NonEmptyString(),
+ MEDIUM,
+ REMOTE_STORAGE_MANAGER_CONFIG_PREFIX_DOC)
+ .define(REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_PROP,
+ STRING,
+ DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX,
+ new ConfigDef.NonEmptyString(),
+ MEDIUM,
+ REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX_DOC)
+ .define(REMOTE_STORAGE_MANAGER_CLASS_NAME_PROP, STRING,
+ null,
+ new ConfigDef.NonEmptyString(),
+ MEDIUM,
+ REMOTE_STORAGE_MANAGER_CLASS_NAME_DOC)
+ .define(REMOTE_STORAGE_MANAGER_CLASS_PATH_PROP, STRING,
+ null,
+ null,
+ MEDIUM,
+ REMOTE_STORAGE_MANAGER_CLASS_PATH_DOC)
+ .define(REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP,
+ STRING,
+ DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME,
+ new ConfigDef.NonEmptyString(),
+ MEDIUM,
+ REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_DOC)
+ .define(REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_PROP,
+ STRING,
+ null,
+ null,
+ MEDIUM,
+ REMOTE_LOG_METADATA_MANAGER_CLASS_PATH_DOC)
+ .define(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP, STRING,
+ null,
+ new ConfigDef.NonEmptyString(),
+ MEDIUM,
+ REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_DOC)
+ .define(REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES,
+ atLeast(0),
+ LOW,
+ REMOTE_LOG_METADATA_CUSTOM_METADATA_MAX_BYTES_DOC)
+ .define(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP,
+ LONG,
+ DEFAULT_REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES,
+ atLeast(1),
+ LOW,
+ REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_DOC)
+ .define(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC)
+ .defineInternal(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
INT,
DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE,
atLeast(1),
MEDIUM,
REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_DOC)
- .defineInternal(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ .defineInternal(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
INT,
DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE,
atLeast(1),
MEDIUM,
REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_DOC)
- .define(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP,
- LONG,
- DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS,
- atLeast(1),
- LOW,
- REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC)
- .defineInternal(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP,
- LONG,
- DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS,
- atLeast(1),
- LOW,
- REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_DOC)
- .defineInternal(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP,
- LONG,
- DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS,
- atLeast(1), LOW,
- REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_DOC)
- .defineInternal(REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP,
- DOUBLE,
- DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_JITTER,
- between(0, 0.5),
- LOW,
- REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_DOC)
- .define(REMOTE_LOG_READER_THREADS_PROP,
- INT,
- DEFAULT_REMOTE_LOG_READER_THREADS,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_READER_THREADS_DOC)
- .define(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP,
- INT,
- DEFAULT_REMOTE_LOG_READER_MAX_PENDING_TASKS,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_READER_MAX_PENDING_TASKS_DOC)
- .define(LOG_LOCAL_RETENTION_MS_PROP,
- LONG,
- DEFAULT_LOG_LOCAL_RETENTION_MS,
- atLeast(DEFAULT_LOG_LOCAL_RETENTION_MS),
- MEDIUM,
- LOG_LOCAL_RETENTION_MS_DOC)
- .define(LOG_LOCAL_RETENTION_BYTES_PROP,
- LONG,
- DEFAULT_LOG_LOCAL_RETENTION_BYTES,
- atLeast(DEFAULT_LOG_LOCAL_RETENTION_BYTES),
- MEDIUM,
- LOG_LOCAL_RETENTION_BYTES_DOC)
- .define(REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP,
- LONG,
- DEFAULT_REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_DOC)
- .define(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_PROP,
- INT,
- DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_DOC)
- .define(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_PROP,
- INT,
- DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_DOC)
- .define(REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP,
- LONG,
- DEFAULT_REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_DOC)
- .define(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP,
- INT,
- DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_DOC)
- .define(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP,
- INT,
- DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS,
- atLeast(1),
- MEDIUM,
- REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_DOC);
+ .define(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP,
+ LONG,
+ DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS,
+ atLeast(1),
+ LOW,
+ REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC)
+ .defineInternal(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP,
+ LONG,
+ DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS,
+ atLeast(1),
+ LOW,
+ REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_DOC)
+ .defineInternal(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP,
+ LONG,
+ DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS,
+ atLeast(1), LOW,
+ REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_DOC)
+ .defineInternal(REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_PROP,
+ DOUBLE,
+ DEFAULT_REMOTE_LOG_MANAGER_TASK_RETRY_JITTER,
+ between(0, 0.5),
+ LOW,
+ REMOTE_LOG_MANAGER_TASK_RETRY_JITTER_DOC)
+ .define(REMOTE_LOG_READER_THREADS_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_READER_THREADS,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_READER_THREADS_DOC)
+ .define(REMOTE_LOG_READER_MAX_PENDING_TASKS_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_READER_MAX_PENDING_TASKS,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_READER_MAX_PENDING_TASKS_DOC)
+ .define(LOG_LOCAL_RETENTION_MS_PROP,
+ LONG,
+ DEFAULT_LOG_LOCAL_RETENTION_MS,
+ atLeast(DEFAULT_LOG_LOCAL_RETENTION_MS),
+ MEDIUM,
+ LOG_LOCAL_RETENTION_MS_DOC)
+ .define(LOG_LOCAL_RETENTION_BYTES_PROP,
+ LONG,
+ DEFAULT_LOG_LOCAL_RETENTION_BYTES,
+ atLeast(DEFAULT_LOG_LOCAL_RETENTION_BYTES),
+ MEDIUM,
+ LOG_LOCAL_RETENTION_BYTES_DOC)
+ .define(REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_PROP,
+ LONG,
+ DEFAULT_REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND_DOC)
+ .define(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM_DOC)
+ .define(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_DOC)
+ .define(REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP,
+ LONG,
+ DEFAULT_REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_DOC)
+ .define(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_DOC)
+ .define(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_DOC)
+ .define(REMOTE_FETCH_MAX_WAIT_MS_PROP,
+ INT,
+ DEFAULT_REMOTE_FETCH_MAX_WAIT_MS,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_FETCH_MAX_WAIT_MS_DOC);
}
private final boolean enableRemoteStorageSystem;
@@ -375,6 +385,7 @@ public final class RemoteLogManagerConfig {
private final long remoteLogManagerFetchMaxBytesPerSecond;
private final int remoteLogManagerFetchNumQuotaSamples;
private final int remoteLogManagerFetchQuotaWindowSizeSeconds;
+ private final int remoteFetchMaxWaitMs;
public RemoteLogManagerConfig(AbstractConfig config) {
this(config.getBoolean(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP),
@@ -407,7 +418,8 @@ public final class RemoteLogManagerConfig {
config.getInt(REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS_PROP),
config.getLong(REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP),
config.getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM_PROP),
- config.getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP));
+ config.getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP),
+ config.getInt(REMOTE_FETCH_MAX_WAIT_MS_PROP));
}
// Visible for testing
@@ -437,8 +449,8 @@ public final class RemoteLogManagerConfig {
int remoteLogManagerCopyQuotaWindowSizeSeconds,
long remoteLogManagerFetchMaxBytesPerSecond,
int remoteLogManagerFetchNumQuotaSamples,
- int remoteLogManagerFetchQuotaWindowSizeSeconds
- ) {
+ int remoteLogManagerFetchQuotaWindowSizeSeconds,
+ int remoteFetchMaxWaitMs) {
this.enableRemoteStorageSystem = enableRemoteStorageSystem;
this.remoteStorageManagerClassName = remoteStorageManagerClassName;
this.remoteStorageManagerClassPath = remoteStorageManagerClassPath;
@@ -466,6 +478,7 @@ public final class RemoteLogManagerConfig {
this.remoteLogManagerFetchMaxBytesPerSecond = remoteLogManagerFetchMaxBytesPerSecond;
this.remoteLogManagerFetchNumQuotaSamples = remoteLogManagerFetchNumQuotaSamples;
this.remoteLogManagerFetchQuotaWindowSizeSeconds = remoteLogManagerFetchQuotaWindowSizeSeconds;
+ this.remoteFetchMaxWaitMs = remoteFetchMaxWaitMs;
}
public boolean enableRemoteStorageSystem() {
@@ -576,6 +589,9 @@ public final class RemoteLogManagerConfig {
return remoteLogManagerFetchQuotaWindowSizeSeconds;
}
+ public int remoteFetchMaxWaitMs() {
+ return remoteFetchMaxWaitMs;
+ }
@Override
public boolean equals(Object o) {
@@ -608,20 +624,22 @@ public final class RemoteLogManagerConfig {
&& remoteLogManagerCopyQuotaWindowSizeSeconds == that.remoteLogManagerCopyQuotaWindowSizeSeconds
&& remoteLogManagerFetchMaxBytesPerSecond == that.remoteLogManagerFetchMaxBytesPerSecond
&& remoteLogManagerFetchNumQuotaSamples == that.remoteLogManagerFetchNumQuotaSamples
- && remoteLogManagerFetchQuotaWindowSizeSeconds == that.remoteLogManagerFetchQuotaWindowSizeSeconds;
+ && remoteLogManagerFetchQuotaWindowSizeSeconds == that.remoteLogManagerFetchQuotaWindowSizeSeconds
+ && remoteFetchMaxWaitMs == that.remoteFetchMaxWaitMs;
}
@Override
public int hashCode() {
- return Objects.hash(enableRemoteStorageSystem, remoteStorageManagerClassName, remoteStorageManagerClassPath,
- remoteLogMetadataManagerClassName, remoteLogMetadataManagerClassPath, remoteLogMetadataManagerListenerName,
- remoteLogMetadataCustomMetadataMaxBytes, remoteLogIndexFileCacheTotalSizeBytes, remoteLogManagerThreadPoolSize,
- remoteLogManagerCopierThreadPoolSize, remoteLogManagerExpirationThreadPoolSize, remoteLogManagerTaskIntervalMs,
- remoteLogManagerTaskRetryBackoffMs, remoteLogManagerTaskRetryBackoffMaxMs, remoteLogManagerTaskRetryJitter,
- remoteLogReaderThreads, remoteLogReaderMaxPendingTasks, remoteStorageManagerProps, remoteLogMetadataManagerProps,
- remoteStorageManagerPrefix, remoteLogMetadataManagerPrefix, remoteLogManagerCopyMaxBytesPerSecond,
- remoteLogManagerCopyNumQuotaSamples, remoteLogManagerCopyQuotaWindowSizeSeconds, remoteLogManagerFetchMaxBytesPerSecond,
- remoteLogManagerFetchNumQuotaSamples, remoteLogManagerFetchQuotaWindowSizeSeconds);
+ return Objects.hash(
+ enableRemoteStorageSystem, remoteStorageManagerClassName, remoteStorageManagerClassPath,
+ remoteLogMetadataManagerClassName, remoteLogMetadataManagerClassPath, remoteLogMetadataManagerListenerName,
+ remoteLogMetadataCustomMetadataMaxBytes, remoteLogIndexFileCacheTotalSizeBytes, remoteLogManagerThreadPoolSize,
+ remoteLogManagerCopierThreadPoolSize, remoteLogManagerExpirationThreadPoolSize, remoteLogManagerTaskIntervalMs,
+ remoteLogManagerTaskRetryBackoffMs, remoteLogManagerTaskRetryBackoffMaxMs, remoteLogManagerTaskRetryJitter,
+ remoteLogReaderThreads, remoteLogReaderMaxPendingTasks, remoteStorageManagerProps, remoteLogMetadataManagerProps,
+ remoteStorageManagerPrefix, remoteLogMetadataManagerPrefix, remoteLogManagerCopyMaxBytesPerSecond,
+ remoteLogManagerCopyNumQuotaSamples, remoteLogManagerCopyQuotaWindowSizeSeconds, remoteLogManagerFetchMaxBytesPerSecond,
+ remoteLogManagerFetchNumQuotaSamples, remoteLogManagerFetchQuotaWindowSizeSeconds, remoteFetchMaxWaitMs);
}
public static void main(String[] args) {
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
index d04c42409b2..058f81626d1 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
@@ -95,7 +95,8 @@ public class RemoteLogManagerConfigTest {
1,
Long.MAX_VALUE,
11,
- 1);
+ 1,
+ 500);
}
private Map<String, Object> extractProps(RemoteLogManagerConfig remoteLogManagerConfig) {
@@ -189,7 +190,8 @@ public class RemoteLogManagerConfigTest {
1,
Long.MAX_VALUE,
11,
- 1);
+ 1,
+ 500);
assertNotEquals(config1.hashCode(), config3.hashCode());
assertNotEquals(config1, config3);
(kafka) 05/12: KAFKA-16852 Adding two thread pools kafka-16852 (#16154)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit e4a3da6b09912e9f1d1fce61e7d77978e84e3c8d
Author: Murali Basani <mu...@aiven.io>
AuthorDate: Mon Jun 3 10:52:54 2024 +0200
KAFKA-16852 Adding two thread pools kafka-16852 (#16154)
Reviewers: Christo Lolov <lo...@amazon.com>, Chia-Ping Tasi <ch...@gmail.com>
---
.../log/remote/storage/RemoteLogManagerConfig.java | 38 ++++++++++++++++++++++
.../remote/storage/RemoteLogManagerConfigTest.java | 6 +++-
2 files changed, 43 insertions(+), 1 deletion(-)
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index 6ea752886a9..d6cf615c781 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -100,6 +100,16 @@ public final class RemoteLogManagerConfig {
"segments, fetch remote log indexes and clean up remote log segments.";
public static final int DEFAULT_REMOTE_LOG_MANAGER_THREAD_POOL_SIZE = 10;
+ public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP = "remote.log.manager.copier.thread.pool.size";
+ public static final String REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in " +
+ "scheduling tasks to copy segments.";
+ public static final int DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE = 10;
+
+ public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP = "remote.log.manager.expiration.thread.pool.size";
+ public static final String REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_DOC = "Size of the thread pool used in" +
+ " scheduling tasks to clean up remote log segments.";
+ public static final int DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE = 10;
+
public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP = "remote.log.manager.task.interval.ms";
public static final String REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_DOC = "Interval at which remote log manager runs the scheduled tasks like copy " +
"segments, and clean up remote log segments.";
@@ -241,6 +251,18 @@ public final class RemoteLogManagerConfig {
atLeast(1),
MEDIUM,
REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_DOC)
+ .defineInternal(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_DOC)
+ .defineInternal(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ INT,
+ DEFAULT_REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE,
+ atLeast(1),
+ MEDIUM,
+ REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_DOC)
.define(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP,
LONG,
DEFAULT_REMOTE_LOG_MANAGER_TASK_INTERVAL_MS,
@@ -333,6 +355,8 @@ public final class RemoteLogManagerConfig {
private final String remoteLogMetadataManagerClassPath;
private final long remoteLogIndexFileCacheTotalSizeBytes;
private final int remoteLogManagerThreadPoolSize;
+ private final int remoteLogManagerCopierThreadPoolSize;
+ private final int remoteLogManagerExpirationThreadPoolSize;
private final long remoteLogManagerTaskIntervalMs;
private final long remoteLogManagerTaskRetryBackoffMs;
private final long remoteLogManagerTaskRetryBackoffMaxMs;
@@ -361,6 +385,8 @@ public final class RemoteLogManagerConfig {
config.getString(REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP),
config.getLong(REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP),
config.getInt(REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP),
+ config.getInt(REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP),
+ config.getInt(REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP),
config.getLong(REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP),
config.getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP),
config.getLong(REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MAX_MS_PROP),
@@ -393,6 +419,8 @@ public final class RemoteLogManagerConfig {
String remoteLogMetadataManagerListenerName,
long remoteLogIndexFileCacheTotalSizeBytes,
int remoteLogManagerThreadPoolSize,
+ int remoteLogManagerCopierThreadPoolSize,
+ int remoteLogManagerExpirationThreadPoolSize,
long remoteLogManagerTaskIntervalMs,
long remoteLogManagerTaskRetryBackoffMs,
long remoteLogManagerTaskRetryBackoffMaxMs,
@@ -418,6 +446,8 @@ public final class RemoteLogManagerConfig {
this.remoteLogMetadataManagerClassPath = remoteLogMetadataManagerClassPath;
this.remoteLogIndexFileCacheTotalSizeBytes = remoteLogIndexFileCacheTotalSizeBytes;
this.remoteLogManagerThreadPoolSize = remoteLogManagerThreadPoolSize;
+ this.remoteLogManagerCopierThreadPoolSize = remoteLogManagerCopierThreadPoolSize;
+ this.remoteLogManagerExpirationThreadPoolSize = remoteLogManagerExpirationThreadPoolSize;
this.remoteLogManagerTaskIntervalMs = remoteLogManagerTaskIntervalMs;
this.remoteLogManagerTaskRetryBackoffMs = remoteLogManagerTaskRetryBackoffMs;
this.remoteLogManagerTaskRetryBackoffMaxMs = remoteLogManagerTaskRetryBackoffMaxMs;
@@ -466,6 +496,14 @@ public final class RemoteLogManagerConfig {
return remoteLogManagerThreadPoolSize;
}
+ public int remoteLogManagerCopierThreadPoolSize() {
+ return remoteLogManagerCopierThreadPoolSize;
+ }
+
+ public int remoteLogManagerExpirationThreadPoolSize() {
+ return remoteLogManagerExpirationThreadPoolSize;
+ }
+
public long remoteLogManagerTaskIntervalMs() {
return remoteLogManagerTaskIntervalMs;
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
index 45fd6669e7d..4e3c2fc26cb 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
@@ -47,7 +47,7 @@ public class RemoteLogManagerConfigTest {
RemoteLogManagerConfig expectedRemoteLogManagerConfig
= new RemoteLogManagerConfig(true, "dummy.remote.storage.class", "dummy.remote.storage.class.path",
remoteLogMetadataManagerClass, "dummy.remote.log.metadata.class.path",
- "listener.name", 1024 * 1024L, 1, 60000L, 100L, 60000L, 0.3, 10, 100, 100,
+ "listener.name", 1024 * 1024L, 1, 1, 1, 60000L, 100L, 60000L, 0.3, 10, 100, 100,
rsmPrefix, rsmProps, rlmmPrefix, rlmmProps, Long.MAX_VALUE, 11, 1,
Long.MAX_VALUE, 11, 1);
@@ -81,6 +81,10 @@ public class RemoteLogManagerConfigTest {
remoteLogManagerConfig.remoteLogIndexFileCacheTotalSizeBytes());
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_THREAD_POOL_SIZE_PROP,
remoteLogManagerConfig.remoteLogManagerThreadPoolSize());
+ props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_COPIER_THREAD_POOL_SIZE_PROP,
+ remoteLogManagerConfig.remoteLogManagerCopierThreadPoolSize());
+ props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_EXPIRATION_THREAD_POOL_SIZE_PROP,
+ remoteLogManagerConfig.remoteLogManagerExpirationThreadPoolSize());
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP,
remoteLogManagerConfig.remoteLogManagerTaskIntervalMs());
props.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_RETRY_BACK_OFF_MS_PROP,
(kafka) 06/12: KAFKA-16880 Update equals and hashcode methods for two attributes (#16173)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 944f4699a7de443a72fa2618b5cbc8531f12a8c4
Author: Murali Basani <mu...@aiven.io>
AuthorDate: Tue Jun 4 08:05:00 2024 +0200
KAFKA-16880 Update equals and hashcode methods for two attributes (#16173)
Reviewers: Kamal Chandraprakash <ka...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../log/remote/storage/RemoteLogManagerConfig.java | 5 +-
.../remote/storage/RemoteLogManagerConfigTest.java | 107 +++++++++++++++++++--
2 files changed, 102 insertions(+), 10 deletions(-)
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index d6cf615c781..1add933d788 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -585,6 +585,8 @@ public final class RemoteLogManagerConfig {
return enableRemoteStorageSystem == that.enableRemoteStorageSystem
&& remoteLogIndexFileCacheTotalSizeBytes == that.remoteLogIndexFileCacheTotalSizeBytes
&& remoteLogManagerThreadPoolSize == that.remoteLogManagerThreadPoolSize
+ && remoteLogManagerCopierThreadPoolSize == that.remoteLogManagerCopierThreadPoolSize
+ && remoteLogManagerExpirationThreadPoolSize == that.remoteLogManagerExpirationThreadPoolSize
&& remoteLogManagerTaskIntervalMs == that.remoteLogManagerTaskIntervalMs
&& remoteLogManagerTaskRetryBackoffMs == that.remoteLogManagerTaskRetryBackoffMs
&& remoteLogManagerTaskRetryBackoffMaxMs == that.remoteLogManagerTaskRetryBackoffMaxMs
@@ -613,7 +615,8 @@ public final class RemoteLogManagerConfig {
public int hashCode() {
return Objects.hash(enableRemoteStorageSystem, remoteStorageManagerClassName, remoteStorageManagerClassPath,
remoteLogMetadataManagerClassName, remoteLogMetadataManagerClassPath, remoteLogMetadataManagerListenerName,
- remoteLogMetadataCustomMetadataMaxBytes, remoteLogIndexFileCacheTotalSizeBytes, remoteLogManagerThreadPoolSize, remoteLogManagerTaskIntervalMs,
+ remoteLogMetadataCustomMetadataMaxBytes, remoteLogIndexFileCacheTotalSizeBytes, remoteLogManagerThreadPoolSize,
+ remoteLogManagerCopierThreadPoolSize, remoteLogManagerExpirationThreadPoolSize, remoteLogManagerTaskIntervalMs,
remoteLogManagerTaskRetryBackoffMs, remoteLogManagerTaskRetryBackoffMaxMs, remoteLogManagerTaskRetryJitter,
remoteLogReaderThreads, remoteLogReaderMaxPendingTasks, remoteStorageManagerProps, remoteLogMetadataManagerProps,
remoteStorageManagerPrefix, remoteLogMetadataManagerPrefix, remoteLogManagerCopyMaxBytesPerSecond,
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
index 4e3c2fc26cb..d04c42409b2 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfigTest.java
@@ -17,7 +17,7 @@
package org.apache.kafka.server.log.remote.storage;
import org.apache.kafka.common.config.AbstractConfig;
-import org.junit.jupiter.api.Assertions;
+import org.junit.jupiter.api.Test;
import org.junit.jupiter.params.ParameterizedTest;
import org.junit.jupiter.params.provider.ValueSource;
@@ -27,6 +27,8 @@ import java.util.Map;
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME;
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP;
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertNotEquals;
public class RemoteLogManagerConfigTest {
@@ -43,13 +45,11 @@ public class RemoteLogManagerConfigTest {
String rlmmPrefix = "__custom.rlmm.";
Map<String, Object> rsmProps = Collections.singletonMap("rsm.prop", "val");
Map<String, Object> rlmmProps = Collections.singletonMap("rlmm.prop", "val");
- String remoteLogMetadataManagerClass = useDefaultRemoteLogMetadataManagerClass ? DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME : "dummy.remote.log.metadata.class";
- RemoteLogManagerConfig expectedRemoteLogManagerConfig
- = new RemoteLogManagerConfig(true, "dummy.remote.storage.class", "dummy.remote.storage.class.path",
- remoteLogMetadataManagerClass, "dummy.remote.log.metadata.class.path",
- "listener.name", 1024 * 1024L, 1, 1, 1, 60000L, 100L, 60000L, 0.3, 10, 100, 100,
- rsmPrefix, rsmProps, rlmmPrefix, rlmmProps, Long.MAX_VALUE, 11, 1,
- Long.MAX_VALUE, 11, 1);
+ RemoteLogManagerConfig expectedRemoteLogManagerConfig = getRemoteLogManagerConfig(useDefaultRemoteLogMetadataManagerClass,
+ rsmPrefix,
+ rlmmPrefix,
+ rsmProps,
+ rlmmProps);
Map<String, Object> props = extractProps(expectedRemoteLogManagerConfig);
rsmProps.forEach((k, v) -> props.put(rsmPrefix + k, v));
@@ -60,7 +60,42 @@ public class RemoteLogManagerConfigTest {
}
TestConfig config = new TestConfig(props);
RemoteLogManagerConfig remoteLogManagerConfig = new RemoteLogManagerConfig(config);
- Assertions.assertEquals(expectedRemoteLogManagerConfig, remoteLogManagerConfig);
+ assertEquals(expectedRemoteLogManagerConfig, remoteLogManagerConfig);
+ }
+
+ private static RemoteLogManagerConfig getRemoteLogManagerConfig(boolean useDefaultRemoteLogMetadataManagerClass,
+ String rsmPrefix,
+ String rlmmPrefix,
+ Map<String, Object> rsmProps,
+ Map<String, Object> rlmmProps) {
+ String remoteLogMetadataManagerClass = useDefaultRemoteLogMetadataManagerClass ? DEFAULT_REMOTE_LOG_METADATA_MANAGER_CLASS_NAME : "dummy.remote.log.metadata.class";
+ return new RemoteLogManagerConfig(true,
+ "dummy.remote.storage.class",
+ "dummy.remote.storage.class.path",
+ remoteLogMetadataManagerClass,
+ "dummy.remote.log.metadata.class.path",
+ "listener.name",
+ 1024 * 1024L,
+ 1,
+ 1,
+ 1,
+ 60000L,
+ 100L,
+ 60000L,
+ 0.3,
+ 10,
+ 100,
+ 100,
+ rsmPrefix,
+ rsmProps,
+ rlmmPrefix,
+ rlmmProps,
+ Long.MAX_VALUE,
+ 11,
+ 1,
+ Long.MAX_VALUE,
+ 11,
+ 1);
}
private Map<String, Object> extractProps(RemoteLogManagerConfig remoteLogManagerConfig) {
@@ -105,4 +140,58 @@ public class RemoteLogManagerConfigTest {
remoteLogManagerConfig.remoteLogMetadataManagerPrefix());
return props;
}
+
+ @Test
+ public void testHashCodeAndEquals_ForAllAndTwoFields() {
+ String rsmPrefix = "__custom.rsm.";
+ String rlmmPrefix = "__custom.rlmm.";
+ Map<String, Object> rsmProps = Collections.singletonMap("rsm.prop", "val");
+ Map<String, Object> rlmmProps = Collections.singletonMap("rlmm.prop", "val");
+ RemoteLogManagerConfig config1 = getRemoteLogManagerConfig(false,
+ rsmPrefix,
+ rlmmPrefix,
+ rsmProps,
+ rlmmProps);
+ RemoteLogManagerConfig config2 = getRemoteLogManagerConfig(false,
+ rsmPrefix,
+ rlmmPrefix,
+ rsmProps,
+ rlmmProps);
+
+ // Initially, hash codes should be equal for default objects
+ assertEquals(config1.hashCode(), config2.hashCode());
+
+ // Initially, objects should be equal
+ assertEquals(config1, config2);
+
+ // Test for specific field remoteLogManagerCopierThreadPoolSize
+ RemoteLogManagerConfig config3 = new RemoteLogManagerConfig(true, "dummy.remote.storage.class",
+ "dummy.remote.storage.class.path",
+ "dummy.remote.log.metadata.class", "dummy.remote.log.metadata.class.path",
+ "listener.name",
+ 1024 * 1024L,
+ 1,
+ 2, // Change here
+ 2, // Change here
+ 60000L,
+ 100L,
+ 60000L,
+ 0.3,
+ 10,
+ 100,
+ 100,
+ rsmPrefix,
+ rsmProps,
+ rlmmPrefix,
+ rlmmProps,
+ Long.MAX_VALUE,
+ 11,
+ 1,
+ Long.MAX_VALUE,
+ 11,
+ 1);
+
+ assertNotEquals(config1.hashCode(), config3.hashCode());
+ assertNotEquals(config1, config3);
+ }
}
(kafka) 07/12: KAFKA-16882 Migrate RemoteLogSegmentLifecycleTest to ClusterInstance infra (#16180)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 69158f67f842da7a166fad5a69bc09e41e9f4e25
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Wed Jun 5 00:41:30 2024 +0530
KAFKA-16882 Migrate RemoteLogSegmentLifecycleTest to ClusterInstance infra (#16180)
- Removed the RemoteLogSegmentLifecycleManager
- Removed the TopicBasedRemoteLogMetadataManagerWrapper, RemoteLogMetadataCacheWrapper, TopicBasedRemoteLogMetadataManagerHarness and TopicBasedRemoteLogMetadataManagerWrapperWithHarness
Reviewers: Kuan-Po (Cooper) Tseng <br...@gmail.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../storage/RemoteLogMetadataManagerTestUtils.java | 23 +-
.../storage/RemoteLogSegmentLifecycleManager.java | 63 ---
.../storage/RemoteLogSegmentLifecycleTest.java | 562 ++++++++-------------
.../TopicBasedRemoteLogMetadataManagerHarness.java | 90 ----
...icBasedRemoteLogMetadataManagerRestartTest.java | 1 -
...RemoteLogMetadataManagerWrapperWithHarness.java | 105 ----
6 files changed, 201 insertions(+), 643 deletions(-)
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataManagerTestUtils.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataManagerTestUtils.java
index fd35167d02a..3fc0432e773 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataManagerTestUtils.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogMetadataManagerTestUtils.java
@@ -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;
@@ -37,10 +32,9 @@ import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemo
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;
@@ -53,7 +47,6 @@ public class RemoteLogMetadataManagerTestUtils {
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;
@@ -85,11 +78,6 @@ public class RemoteLogMetadataManagerTestUtils {
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();
@@ -105,19 +93,12 @@ public class RemoteLogMetadataManagerTestUtils {
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;
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleManager.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleManager.java
deleted file mode 100644
index b8870595152..00000000000
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleManager.java
+++ /dev/null
@@ -1,63 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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 org.apache.kafka.server.log.remote.metadata.storage;
-
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
-import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
-
-import java.io.Closeable;
-import java.io.IOException;
-import java.util.Iterator;
-import java.util.Optional;
-
-/**
- * This interface defines the lifecycle methods for {@code RemoteLogSegmentMetadata}. {@link RemoteLogSegmentLifecycleTest} tests
- * different implementations of this interface. This is responsible for managing all the segments for a given {@code topicIdPartition}
- * registered with {@link #initialize(TopicIdPartition)}.
- *
- * @see org.apache.kafka.server.log.remote.metadata.storage.RemoteLogSegmentLifecycleTest.RemoteLogMetadataCacheWrapper
- * @see org.apache.kafka.server.log.remote.metadata.storage.RemoteLogSegmentLifecycleTest.TopicBasedRemoteLogMetadataManagerWrapper
- */
-public interface RemoteLogSegmentLifecycleManager extends Closeable {
-
- /**
- * Initialize the resources for this instance and register the given {@code topicIdPartition}.
- *
- * @param topicIdPartition topic partition to be registered with this instance.
- */
- default void initialize(TopicIdPartition topicIdPartition) {
- }
-
- @Override
- default void close() throws IOException {
- }
-
- void addRemoteLogSegmentMetadata(RemoteLogSegmentMetadata segmentMetadata) throws RemoteStorageException;
-
- void updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate segmentMetadataUpdate) throws RemoteStorageException;
-
- Optional<Long> highestOffsetForEpoch(int epoch) throws RemoteStorageException;
-
- Optional<RemoteLogSegmentMetadata> remoteLogSegmentMetadata(int leaderEpoch,
- long offset) throws RemoteStorageException;
-
- Iterator<RemoteLogSegmentMetadata> listRemoteLogSegments(int leaderEpoch) throws RemoteStorageException;
-
- Iterator<RemoteLogSegmentMetadata> listAllRemoteLogSegments() throws RemoteStorageException;
-}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleTest.java
index 95fbed72b17..7dbf924653e 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/RemoteLogSegmentLifecycleTest.java
@@ -16,116 +16,132 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;
+import kafka.test.ClusterInstance;
+import kafka.test.annotation.ClusterTest;
+import kafka.test.annotation.ClusterTestDefaults;
+import kafka.test.junit.ClusterTestExtensions;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
-import org.apache.kafka.common.utils.MockTime;
+import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Time;
-import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.server.log.remote.storage.RemoteLogMetadataManager;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState;
-import org.apache.kafka.server.log.remote.storage.RemoteResourceNotFoundException;
import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
+import org.apache.kafka.storage.internals.log.EpochEntry;
import org.apache.kafka.test.TestUtils;
-import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.params.ParameterizedTest;
-import org.junit.jupiter.params.provider.Arguments;
-import org.junit.jupiter.params.provider.MethodSource;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-import java.io.IOException;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.extension.ExtendWith;
+
import java.util.Arrays;
-import java.util.Collection;
import java.util.Collections;
import java.util.HashMap;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
-import java.util.Objects;
import java.util.Optional;
-import java.util.Set;
-import java.util.stream.Stream;
+import java.util.concurrent.ExecutionException;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertFalse;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+import static org.mockito.Mockito.spy;
+import static org.mockito.Mockito.verify;
+
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.COPY_SEGMENT_FINISHED;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.DELETE_SEGMENT_STARTED;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogSegmentState.DELETE_SEGMENT_FINISHED;
+@ClusterTestDefaults(brokers = 3)
+@ExtendWith(value = ClusterTestExtensions.class)
+@Tag("integration")
public class RemoteLogSegmentLifecycleTest {
- private static final Logger log = LoggerFactory.getLogger(RemoteLogSegmentLifecycleTest.class);
- private static final int SEG_SIZE = 1024 * 1024;
- private static final int BROKER_ID_0 = 0;
- private static final int BROKER_ID_1 = 1;
+ private final int segSize = 1048576;
+ private final int brokerId0 = 0;
+ private final int brokerId1 = 1;
+ private final Uuid topicId = Uuid.randomUuid();
+ private final TopicPartition tp = new TopicPartition("foo", 0);
+ private final TopicIdPartition topicIdPartition = new TopicIdPartition(topicId, tp);
+ private final Time time = new SystemTime();
+ private final RemotePartitionMetadataStore spyRemotePartitionMetadataStore = spy(new RemotePartitionMetadataStore());
+ private final ClusterInstance clusterInstance;
+
+ RemoteLogSegmentLifecycleTest(ClusterInstance clusterInstance) { // Constructor injections
+ this.clusterInstance = clusterInstance;
+ }
- private final TopicIdPartition topicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition("foo", 0));
- private final Time time = new MockTime(1);
+ private RemoteLogMetadataManager createTopicBasedRemoteLogMetadataManager() {
+ return RemoteLogMetadataManagerTestUtils.builder()
+ .bootstrapServers(clusterInstance.bootstrapServers())
+ .startConsumerThread(true)
+ .remotePartitionMetadataStore(() -> spyRemotePartitionMetadataStore)
+ .build();
+ }
- @ParameterizedTest(name = "remoteLogSegmentLifecycleManager = {0}")
- @MethodSource("remoteLogSegmentLifecycleManagers")
- public void testRemoteLogSegmentLifeCycle(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager) throws Exception {
- try {
- remoteLogSegmentLifecycleManager.initialize(topicIdPartition);
+ @ClusterTest
+ public void testRemoteLogSegmentLifeCycle() throws Exception {
+ try (RemoteLogMetadataManager metadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ metadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition), Collections.emptySet());
// segment 0
// offsets: [0-100]
// leader epochs (0,0), (1,20), (2,80)
- Map<Integer, Long> segment0LeaderEpochs = new HashMap<>();
- segment0LeaderEpochs.put(0, 0L);
- segment0LeaderEpochs.put(1, 20L);
- segment0LeaderEpochs.put(2, 80L);
- RemoteLogSegmentId segment0Id = new RemoteLogSegmentId(topicIdPartition, Uuid.randomUuid());
- RemoteLogSegmentMetadata segment0Metadata = new RemoteLogSegmentMetadata(segment0Id, 0L, 100L,
- -1L, BROKER_ID_0, time.milliseconds(), SEG_SIZE,
- segment0LeaderEpochs);
- remoteLogSegmentLifecycleManager.addRemoteLogSegmentMetadata(segment0Metadata);
+ Map<Integer, Long> leaderEpochSegment0 = new HashMap<>();
+ leaderEpochSegment0.put(0, 0L);
+ leaderEpochSegment0.put(1, 20L);
+ leaderEpochSegment0.put(2, 80L);
+ RemoteLogSegmentId segmentId0 = new RemoteLogSegmentId(topicIdPartition, Uuid.randomUuid());
+ RemoteLogSegmentMetadata metadataSegment0 = new RemoteLogSegmentMetadata(segmentId0, 0L,
+ 100L, -1L, brokerId0, time.milliseconds(), segSize, leaderEpochSegment0);
+ metadataManager.addRemoteLogSegmentMetadata(metadataSegment0).get();
+ verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadata(metadataSegment0);
// We should not get this as the segment is still getting copied and it is not yet considered successful until
- // it reaches RemoteLogSegmentState.COPY_SEGMENT_FINISHED.
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(40, 1).isPresent());
+ // it reaches COPY_SEGMENT_FINISHED.
+ assertFalse(metadataManager.remoteLogSegmentMetadata(topicIdPartition, 1, 40).isPresent());
// Check that these leader epochs are not to be considered for highestOffsetForEpoch API as they are still getting copied.
- Stream.of(0, 1, 2).forEach(epoch -> {
- try {
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.highestOffsetForEpoch(epoch).isPresent());
- } catch (RemoteStorageException e) {
- Assertions.fail(e);
- }
- });
+ for (int leaderEpoch = 0; leaderEpoch <= 2; leaderEpoch++) {
+ assertFalse(metadataManager.highestOffsetForEpoch(topicIdPartition, leaderEpoch).isPresent());
+ }
- RemoteLogSegmentMetadataUpdate segment0Update = new RemoteLogSegmentMetadataUpdate(
- segment0Id, time.milliseconds(), Optional.empty(),
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED, BROKER_ID_1);
- remoteLogSegmentLifecycleManager.updateRemoteLogSegmentMetadata(segment0Update);
- RemoteLogSegmentMetadata expectedSegment0Metadata = segment0Metadata.createWithUpdates(segment0Update);
+ RemoteLogSegmentMetadataUpdate metadataUpdateSegment0 = new RemoteLogSegmentMetadataUpdate(
+ segmentId0, time.milliseconds(), Optional.empty(),
+ COPY_SEGMENT_FINISHED, brokerId1);
+ metadataManager.updateRemoteLogSegmentMetadata(metadataUpdateSegment0).get();
+ verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadataUpdate(metadataUpdateSegment0);
+ metadataSegment0 = metadataSegment0.createWithUpdates(metadataUpdateSegment0);
// segment 1
// offsets: [101 - 200]
// no changes in leadership with in this segment
// leader epochs (2, 101)
- Map<Integer, Long> segment1LeaderEpochs = Collections.singletonMap(2, 101L);
- RemoteLogSegmentMetadata segment1Metadata = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager, segment1LeaderEpochs, 101L,
- 200L,
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ Map<Integer, Long> leaderEpochSegment1 = Collections.singletonMap(2, 101L);
+ RemoteLogSegmentMetadata metadataSegment1 = upsertSegmentState(metadataManager, leaderEpochSegment1,
+ 101L, 200L, COPY_SEGMENT_FINISHED);
// segment 2
// offsets: [201 - 300]
// moved to epoch 3 in between
// leader epochs (2, 201), (3, 240)
- Map<Integer, Long> segment2LeaderEpochs = new HashMap<>();
- segment2LeaderEpochs.put(2, 201L);
- segment2LeaderEpochs.put(3, 240L);
- RemoteLogSegmentMetadata segment2Metadata = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager, segment2LeaderEpochs, 201L,
- 300L,
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ Map<Integer, Long> leaderEpochSegment2 = new HashMap<>();
+ leaderEpochSegment2.put(2, 201L);
+ leaderEpochSegment2.put(3, 240L);
+ RemoteLogSegmentMetadata metadataSegment2 = upsertSegmentState(metadataManager, leaderEpochSegment2,
+ 201L, 300L, COPY_SEGMENT_FINISHED);
// segment 3
// offsets: [250 - 400]
// leader epochs (3, 250), (4, 370)
- Map<Integer, Long> segment3LeaderEpochs = new HashMap<>();
- segment3LeaderEpochs.put(3, 250L);
- segment3LeaderEpochs.put(4, 370L);
- RemoteLogSegmentMetadata segment3Metadata = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager, segment3LeaderEpochs, 250L,
- 400L,
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ Map<Integer, Long> leaderEpochSegment3 = new HashMap<>();
+ leaderEpochSegment3.put(3, 250L);
+ leaderEpochSegment3.put(4, 370L);
+ RemoteLogSegmentMetadata metadataSegment3 = upsertSegmentState(metadataManager, leaderEpochSegment3,
+ 250L, 400L, COPY_SEGMENT_FINISHED);
//////////////////////////////////////////////////////////////////////////////////////////
// Four segments are added with different boundaries and leader epochs.
@@ -133,55 +149,49 @@ public class RemoteLogSegmentLifecycleTest {
// epochs and offsets
//////////////////////////////////////////////////////////////////////////////////////////
- HashMap<EpochOffset, RemoteLogSegmentMetadata> expectedEpochOffsetToSegmentMetadata = new HashMap<>();
+ Map<EpochEntry, RemoteLogSegmentMetadata> expectedEpochEntryToMetadata = new HashMap<>();
// Existing metadata entries.
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(1, 40), expectedSegment0Metadata);
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(2, 110), segment1Metadata);
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(3, 240), segment2Metadata);
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(3, 250), segment3Metadata);
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(4, 375), segment3Metadata);
+ expectedEpochEntryToMetadata.put(new EpochEntry(1, 40), metadataSegment0);
+ expectedEpochEntryToMetadata.put(new EpochEntry(2, 110), metadataSegment1);
+ expectedEpochEntryToMetadata.put(new EpochEntry(3, 240), metadataSegment2);
+ expectedEpochEntryToMetadata.put(new EpochEntry(3, 250), metadataSegment3);
+ expectedEpochEntryToMetadata.put(new EpochEntry(4, 375), metadataSegment3);
// Non existing metadata entries.
// Search for offset 110, epoch 1, and it should not exist.
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(1, 110), null);
+ expectedEpochEntryToMetadata.put(new EpochEntry(1, 110), null);
// Search for non existing offset 401, epoch 4.
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(4, 401), null);
+ expectedEpochEntryToMetadata.put(new EpochEntry(4, 401), null);
// Search for non existing epoch 5.
- expectedEpochOffsetToSegmentMetadata.put(new EpochOffset(5, 301), null);
+ expectedEpochEntryToMetadata.put(new EpochEntry(5, 301), null);
- for (Map.Entry<EpochOffset, RemoteLogSegmentMetadata> entry : expectedEpochOffsetToSegmentMetadata.entrySet()) {
- EpochOffset epochOffset = entry.getKey();
- Optional<RemoteLogSegmentMetadata> segmentMetadata = remoteLogSegmentLifecycleManager
- .remoteLogSegmentMetadata(epochOffset.epoch, epochOffset.offset);
+ for (Map.Entry<EpochEntry, RemoteLogSegmentMetadata> entry : expectedEpochEntryToMetadata.entrySet()) {
+ EpochEntry epochEntry = entry.getKey();
+ Optional<RemoteLogSegmentMetadata> actualMetadataOpt = metadataManager
+ .remoteLogSegmentMetadata(topicIdPartition, epochEntry.epoch, epochEntry.startOffset);
RemoteLogSegmentMetadata expectedSegmentMetadata = entry.getValue();
- log.debug("Searching for {} , result: {}, expected: {} ", epochOffset, segmentMetadata,
- expectedSegmentMetadata);
if (expectedSegmentMetadata != null) {
- Assertions.assertEquals(Optional.of(expectedSegmentMetadata), segmentMetadata);
+ assertEquals(Optional.of(expectedSegmentMetadata), actualMetadataOpt);
} else {
- Assertions.assertFalse(segmentMetadata.isPresent());
+ assertFalse(actualMetadataOpt.isPresent());
}
}
// Update segment with state as DELETE_SEGMENT_STARTED.
// It should not be available when we search for that segment.
- remoteLogSegmentLifecycleManager
- .updateRemoteLogSegmentMetadata(new RemoteLogSegmentMetadataUpdate(expectedSegment0Metadata.remoteLogSegmentId(),
- time.milliseconds(),
- Optional.empty(),
- RemoteLogSegmentState.DELETE_SEGMENT_STARTED,
- BROKER_ID_1));
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(0, 10).isPresent());
+ RemoteLogSegmentMetadataUpdate metadataDeleteStartedSegment0 =
+ new RemoteLogSegmentMetadataUpdate(metadataSegment0.remoteLogSegmentId(), time.milliseconds(),
+ Optional.empty(), DELETE_SEGMENT_STARTED, brokerId1);
+ metadataManager.updateRemoteLogSegmentMetadata(metadataDeleteStartedSegment0).get();
+ assertFalse(metadataManager.remoteLogSegmentMetadata(topicIdPartition, 0, 10).isPresent());
// Update segment with state as DELETE_SEGMENT_FINISHED.
// It should not be available when we search for that segment.
- remoteLogSegmentLifecycleManager
- .updateRemoteLogSegmentMetadata(new RemoteLogSegmentMetadataUpdate(expectedSegment0Metadata.remoteLogSegmentId(),
- time.milliseconds(),
- Optional.empty(),
- RemoteLogSegmentState.DELETE_SEGMENT_FINISHED,
- BROKER_ID_1));
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(0, 10).isPresent());
+ RemoteLogSegmentMetadataUpdate metadataDeleteFinishedSegment0 =
+ new RemoteLogSegmentMetadataUpdate(metadataSegment0.remoteLogSegmentId(), time.milliseconds(),
+ Optional.empty(), DELETE_SEGMENT_FINISHED, brokerId1);
+ metadataManager.updateRemoteLogSegmentMetadata(metadataDeleteFinishedSegment0).get();
+ assertFalse(metadataManager.remoteLogSegmentMetadata(topicIdPartition, 0, 10).isPresent());
//////////////////////////////////////////////////////////////////////////////////////////
// Search for cache.highestLogOffset(leaderEpoch) for all the leader epochs
@@ -197,333 +207,159 @@ public class RemoteLogSegmentLifecycleTest {
for (Map.Entry<Integer, Long> entry : expectedEpochToHighestOffset.entrySet()) {
Integer epoch = entry.getKey();
Long expectedOffset = entry.getValue();
- Optional<Long> offset = remoteLogSegmentLifecycleManager.highestOffsetForEpoch(epoch);
- log.debug("Fetching highest offset for epoch: {} , returned: {} , expected: {}", epoch, offset, expectedOffset);
- Assertions.assertEquals(Optional.of(expectedOffset), offset);
+ Optional<Long> offset = metadataManager.highestOffsetForEpoch(topicIdPartition, epoch);
+ assertEquals(Optional.of(expectedOffset), offset);
}
// Search for non existing leader epoch
- Optional<Long> highestOffsetForEpoch5 = remoteLogSegmentLifecycleManager.highestOffsetForEpoch(5);
- Assertions.assertFalse(highestOffsetForEpoch5.isPresent());
- } finally {
- Utils.closeQuietly(remoteLogSegmentLifecycleManager, "RemoteLogSegmentLifecycleManager");
+ Optional<Long> highestOffsetForEpoch5 = metadataManager.highestOffsetForEpoch(topicIdPartition, 5);
+ assertFalse(highestOffsetForEpoch5.isPresent());
}
}
- private RemoteLogSegmentMetadata createSegmentUpdateWithState(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager,
- Map<Integer, Long> segmentLeaderEpochs,
- long startOffset,
- long endOffset,
- RemoteLogSegmentState state)
- throws RemoteStorageException {
+ private RemoteLogSegmentMetadata upsertSegmentState(RemoteLogMetadataManager metadataManager,
+ Map<Integer, Long> segmentLeaderEpochs,
+ long startOffset,
+ long endOffset,
+ RemoteLogSegmentState state)
+ throws RemoteStorageException, ExecutionException, InterruptedException {
RemoteLogSegmentId segmentId = new RemoteLogSegmentId(topicIdPartition, Uuid.randomUuid());
- RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(segmentId, startOffset, endOffset, -1L, BROKER_ID_0,
- time.milliseconds(), SEG_SIZE, segmentLeaderEpochs);
- remoteLogSegmentLifecycleManager.addRemoteLogSegmentMetadata(segmentMetadata);
-
- RemoteLogSegmentMetadataUpdate segMetadataUpdate = new RemoteLogSegmentMetadataUpdate(segmentId, time.milliseconds(),
- Optional.empty(),
- state, BROKER_ID_1);
- remoteLogSegmentLifecycleManager.updateRemoteLogSegmentMetadata(segMetadataUpdate);
-
+ RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(segmentId, startOffset, endOffset,
+ -1L, brokerId0, time.milliseconds(), segSize, segmentLeaderEpochs);
+ metadataManager.addRemoteLogSegmentMetadata(segmentMetadata).get();
+ verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadata(segmentMetadata);
+
+ RemoteLogSegmentMetadataUpdate segMetadataUpdate = new RemoteLogSegmentMetadataUpdate(segmentId,
+ time.milliseconds(), Optional.empty(), state, brokerId1);
+ metadataManager.updateRemoteLogSegmentMetadata(segMetadataUpdate).get();
+ verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadataUpdate(segMetadataUpdate);
return segmentMetadata.createWithUpdates(segMetadataUpdate);
}
- private static class EpochOffset {
- final int epoch;
- final long offset;
-
- private EpochOffset(int epoch,
- long offset) {
- this.epoch = epoch;
- this.offset = offset;
- }
-
- @Override
- public boolean equals(Object o) {
- if (this == o) {
- return true;
- }
- if (o == null || getClass() != o.getClass()) {
- return false;
- }
- EpochOffset that = (EpochOffset) o;
- return epoch == that.epoch && offset == that.offset;
- }
-
- @Override
- public int hashCode() {
- return Objects.hash(epoch, offset);
- }
-
- @Override
- public String toString() {
- return "EpochOffset{" +
- "epoch=" + epoch +
- ", offset=" + offset +
- '}';
- }
- }
-
- private static Collection<Arguments> remoteLogSegmentLifecycleManagers() {
- return Arrays.asList(Arguments.of(new RemoteLogMetadataCacheWrapper()),
- Arguments.of(new TopicBasedRemoteLogMetadataManagerWrapper()));
- }
-
- private void checkListSegments(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager,
+ private void checkListSegments(RemoteLogMetadataManager metadataManager,
int leaderEpoch,
- RemoteLogSegmentMetadata expectedSegment)
+ RemoteLogSegmentMetadata expectedMetadata)
throws RemoteStorageException {
// cache.listRemoteLogSegments(leaderEpoch) should contain the above segment.
- Iterator<RemoteLogSegmentMetadata> segmentsIter = remoteLogSegmentLifecycleManager.listRemoteLogSegments(leaderEpoch);
- Assertions.assertTrue(segmentsIter.hasNext());
- Assertions.assertEquals(expectedSegment, segmentsIter.next());
+ Iterator<RemoteLogSegmentMetadata> metadataIter =
+ metadataManager.listRemoteLogSegments(topicIdPartition, leaderEpoch);
+ assertTrue(metadataIter.hasNext());
+ assertEquals(expectedMetadata, metadataIter.next());
// cache.listAllRemoteLogSegments() should contain the above segment.
- Iterator<RemoteLogSegmentMetadata> allSegmentsIter = remoteLogSegmentLifecycleManager.listAllRemoteLogSegments();
- Assertions.assertTrue(allSegmentsIter.hasNext());
- Assertions.assertEquals(expectedSegment, allSegmentsIter.next());
+ Iterator<RemoteLogSegmentMetadata> allMetadataIter = metadataManager.listRemoteLogSegments(topicIdPartition);
+ assertTrue(allMetadataIter.hasNext());
+ assertEquals(expectedMetadata, allMetadataIter.next());
}
- @ParameterizedTest(name = "remoteLogSegmentLifecycleManager = {0}")
- @MethodSource("remoteLogSegmentLifecycleManagers")
- public void testCacheSegmentWithCopySegmentStartedState(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager) throws Exception {
-
- try {
- remoteLogSegmentLifecycleManager.initialize(topicIdPartition);
-
+ @ClusterTest
+ public void testCacheSegmentWithCopySegmentStartedState() throws Exception {
+ try (RemoteLogMetadataManager metadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ metadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition), Collections.emptySet());
// Create a segment with state COPY_SEGMENT_STARTED, and check for searching that segment and listing the
// segments.
RemoteLogSegmentId segmentId = new RemoteLogSegmentId(topicIdPartition, Uuid.randomUuid());
- RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(segmentId, 0L, 50L, -1L, BROKER_ID_0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- remoteLogSegmentLifecycleManager.addRemoteLogSegmentMetadata(segmentMetadata);
+ RemoteLogSegmentMetadata segmentMetadata = new RemoteLogSegmentMetadata(segmentId, 0L, 50L,
+ -1L, brokerId0, time.milliseconds(), segSize, Collections.singletonMap(0, 0L));
+ metadataManager.addRemoteLogSegmentMetadata(segmentMetadata).get();
+ verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadata(segmentMetadata);
// This segment should not be available as the state is not reached to COPY_SEGMENT_FINISHED.
- Optional<RemoteLogSegmentMetadata> segMetadataForOffset0Epoch0 = remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(0, 0);
- Assertions.assertFalse(segMetadataForOffset0Epoch0.isPresent());
+ Optional<RemoteLogSegmentMetadata> segMetadataForOffset0Epoch0 =
+ metadataManager.remoteLogSegmentMetadata(topicIdPartition, 0, 0);
+ assertFalse(segMetadataForOffset0Epoch0.isPresent());
// cache.listRemoteLogSegments APIs should contain the above segment.
- checkListSegments(remoteLogSegmentLifecycleManager, 0, segmentMetadata);
- } finally {
- Utils.closeQuietly(remoteLogSegmentLifecycleManager, "RemoteLogSegmentLifecycleManager");
+ checkListSegments(metadataManager, 0, segmentMetadata);
}
}
- @ParameterizedTest(name = "remoteLogSegmentLifecycleManager = {0}")
- @MethodSource("remoteLogSegmentLifecycleManagers")
- public void testCacheSegmentWithCopySegmentFinishedState(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager) throws Exception {
- try {
- remoteLogSegmentLifecycleManager.initialize(topicIdPartition);
-
+ @ClusterTest
+ public void testCacheSegmentWithCopySegmentFinishedState() throws Exception {
+ try (RemoteLogMetadataManager metadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ metadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition), Collections.emptySet());
// Create a segment and move it to state COPY_SEGMENT_FINISHED. and check for searching that segment and
// listing the segments.
- RemoteLogSegmentMetadata segmentMetadata = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager,
- Collections.singletonMap(0, 101L),
- 101L, 200L, RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ RemoteLogSegmentMetadata segmentMetadata = upsertSegmentState(
+ metadataManager, Collections.singletonMap(0, 101L), 101L, 200L, COPY_SEGMENT_FINISHED);
// Search should return the above segment.
- Optional<RemoteLogSegmentMetadata> segMetadataForOffset150 = remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(0, 150);
- Assertions.assertEquals(Optional.of(segmentMetadata), segMetadataForOffset150);
+ Optional<RemoteLogSegmentMetadata> segMetadataForOffset150 =
+ metadataManager.remoteLogSegmentMetadata(topicIdPartition, 0, 150);
+ assertEquals(Optional.of(segmentMetadata), segMetadataForOffset150);
// cache.listRemoteLogSegments should contain the above segments.
- checkListSegments(remoteLogSegmentLifecycleManager, 0, segmentMetadata);
- } finally {
- Utils.closeQuietly(remoteLogSegmentLifecycleManager, "RemoteLogSegmentLifecycleManager");
+ checkListSegments(metadataManager, 0, segmentMetadata);
}
}
- @ParameterizedTest(name = "remoteLogSegmentLifecycleManager = {0}")
- @MethodSource("remoteLogSegmentLifecycleManagers")
- public void testCacheSegmentWithDeleteSegmentStartedState(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager) throws Exception {
- try {
- remoteLogSegmentLifecycleManager.initialize(topicIdPartition);
-
+ @ClusterTest
+ public void testCacheSegmentWithDeleteSegmentStartedState() throws Exception {
+ try (RemoteLogMetadataManager metadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ metadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition), Collections.emptySet());
// Create a segment and move it to state DELETE_SEGMENT_STARTED, and check for searching that segment and
// listing the segments.
- RemoteLogSegmentMetadata segmentMetadata = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager,
- Collections.singletonMap(0, 201L),
- 201L, 300L, RemoteLogSegmentState.DELETE_SEGMENT_STARTED);
+ RemoteLogSegmentMetadata segmentMetadata = upsertSegmentState(
+ metadataManager, Collections.singletonMap(0, 201L), 201L, 300L, DELETE_SEGMENT_STARTED);
// Search should not return the above segment as their leader epoch state is cleared.
- Optional<RemoteLogSegmentMetadata> segmentMetadataForOffset250Epoch0 = remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(0, 250);
- Assertions.assertFalse(segmentMetadataForOffset250Epoch0.isPresent());
-
- checkListSegments(remoteLogSegmentLifecycleManager, 0, segmentMetadata);
- } finally {
- Utils.closeQuietly(remoteLogSegmentLifecycleManager, "RemoteLogSegmentLifecycleManager");
+ Optional<RemoteLogSegmentMetadata> segmentMetadataForOffset250Epoch0 =
+ metadataManager.remoteLogSegmentMetadata(topicIdPartition, 0, 250);
+ assertFalse(segmentMetadataForOffset250Epoch0.isPresent());
+ checkListSegments(metadataManager, 0, segmentMetadata);
}
}
- @ParameterizedTest(name = "remoteLogSegmentLifecycleManager = {0}")
- @MethodSource("remoteLogSegmentLifecycleManagers")
- public void testCacheSegmentsWithDeleteSegmentFinishedState(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager) throws Exception {
- try {
- remoteLogSegmentLifecycleManager.initialize(topicIdPartition);
-
+ @ClusterTest
+ public void testCacheSegmentsWithDeleteSegmentFinishedState() throws Exception {
+ try (RemoteLogMetadataManager metadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ metadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition), Collections.emptySet());
// Create a segment and move it to state DELETE_SEGMENT_FINISHED, and check for searching that segment and
// listing the segments.
- RemoteLogSegmentMetadata segmentMetadata = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager,
- Collections.singletonMap(0, 301L),
- 301L, 400L, RemoteLogSegmentState.DELETE_SEGMENT_STARTED);
+ RemoteLogSegmentMetadata segmentMetadata = upsertSegmentState(
+ metadataManager, Collections.singletonMap(0, 301L), 301L, 400L, DELETE_SEGMENT_STARTED);
// Search should not return the above segment as their leader epoch state is cleared.
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.remoteLogSegmentMetadata(0, 350).isPresent());
+ assertFalse(metadataManager.remoteLogSegmentMetadata(topicIdPartition, 0, 350).isPresent());
- RemoteLogSegmentMetadataUpdate segmentMetadataUpdate = new RemoteLogSegmentMetadataUpdate(segmentMetadata.remoteLogSegmentId(),
- time.milliseconds(),
- Optional.empty(),
- RemoteLogSegmentState.DELETE_SEGMENT_FINISHED,
- BROKER_ID_1);
- remoteLogSegmentLifecycleManager.updateRemoteLogSegmentMetadata(segmentMetadataUpdate);
+ RemoteLogSegmentMetadataUpdate segmentMetadataUpdate = new RemoteLogSegmentMetadataUpdate(
+ segmentMetadata.remoteLogSegmentId(), time.milliseconds(), Optional.empty(),
+ DELETE_SEGMENT_FINISHED, brokerId1);
+ metadataManager.updateRemoteLogSegmentMetadata(segmentMetadataUpdate).get();
+ verify(spyRemotePartitionMetadataStore).handleRemoteLogSegmentMetadataUpdate(segmentMetadataUpdate);
// listRemoteLogSegments(0) and listRemoteLogSegments() should not contain the above segment.
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.listRemoteLogSegments(0).hasNext());
- Assertions.assertFalse(remoteLogSegmentLifecycleManager.listAllRemoteLogSegments().hasNext());
- } finally {
- Utils.closeQuietly(remoteLogSegmentLifecycleManager, "RemoteLogSegmentLifecycleManager");
+ assertFalse(metadataManager.listRemoteLogSegments(topicIdPartition, 0).hasNext());
+ assertFalse(metadataManager.listRemoteLogSegments(topicIdPartition).hasNext());
}
}
- @ParameterizedTest(name = "remoteLogSegmentLifecycleManager = {0}")
- @MethodSource("remoteLogSegmentLifecycleManagers")
- public void testCacheListSegments(RemoteLogSegmentLifecycleManager remoteLogSegmentLifecycleManager) throws Exception {
- try {
- remoteLogSegmentLifecycleManager.initialize(topicIdPartition);
-
+ @ClusterTest
+ public void testCacheListSegments() throws Exception {
+ try (RemoteLogMetadataManager metadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ metadataManager.onPartitionLeadershipChanges(Collections.singleton(topicIdPartition), Collections.emptySet());
// Create a few segments and add them to the cache.
- RemoteLogSegmentMetadata segment0 = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager, Collections.singletonMap(0, 0L), 0,
- 100,
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
- RemoteLogSegmentMetadata segment1 = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager, Collections.singletonMap(0, 101L), 101,
- 200,
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
- Map<Integer, Long> segment2LeaderEpochs = new HashMap<>();
- segment2LeaderEpochs.put(0, 201L);
- segment2LeaderEpochs.put(1, 301L);
- RemoteLogSegmentMetadata segment2 = createSegmentUpdateWithState(remoteLogSegmentLifecycleManager, segment2LeaderEpochs, 201, 400,
- RemoteLogSegmentState.COPY_SEGMENT_FINISHED);
+ RemoteLogSegmentMetadata segment0 = upsertSegmentState(metadataManager, Collections.singletonMap(0, 0L),
+ 0, 100, COPY_SEGMENT_FINISHED);
+ RemoteLogSegmentMetadata segment1 = upsertSegmentState(metadataManager, Collections.singletonMap(0, 101L),
+ 101, 200, COPY_SEGMENT_FINISHED);
+ Map<Integer, Long> leaderEpochSegment2 = new HashMap<>();
+ leaderEpochSegment2.put(0, 201L);
+ leaderEpochSegment2.put(1, 301L);
+ RemoteLogSegmentMetadata segment2 = upsertSegmentState(metadataManager, leaderEpochSegment2,
+ 201, 400, COPY_SEGMENT_FINISHED);
// listRemoteLogSegments(0) and listAllRemoteLogSegments() should contain all the above segments.
List<RemoteLogSegmentMetadata> expectedSegmentsForEpoch0 = Arrays.asList(segment0, segment1, segment2);
- Assertions.assertTrue(TestUtils.sameElementsWithOrder(remoteLogSegmentLifecycleManager.listRemoteLogSegments(0),
- expectedSegmentsForEpoch0.iterator()));
- Assertions.assertTrue(TestUtils.sameElementsWithoutOrder(remoteLogSegmentLifecycleManager.listAllRemoteLogSegments(),
- expectedSegmentsForEpoch0.iterator()));
+ assertTrue(TestUtils.sameElementsWithOrder(
+ expectedSegmentsForEpoch0.iterator(), metadataManager.listRemoteLogSegments(topicIdPartition, 0)));
+ assertTrue(TestUtils.sameElementsWithoutOrder(
+ expectedSegmentsForEpoch0.iterator(), metadataManager.listRemoteLogSegments(topicIdPartition)));
// listRemoteLogSegments(1) should contain only segment2.
List<RemoteLogSegmentMetadata> expectedSegmentsForEpoch1 = Collections.singletonList(segment2);
- Assertions.assertTrue(TestUtils.sameElementsWithOrder(remoteLogSegmentLifecycleManager.listRemoteLogSegments(1),
- expectedSegmentsForEpoch1.iterator()));
- } finally {
- Utils.closeQuietly(remoteLogSegmentLifecycleManager, "RemoteLogSegmentLifecycleManager");
- }
- }
-
- /**
- * This is a wrapper with {@link TopicBasedRemoteLogMetadataManager} implementing {@link RemoteLogSegmentLifecycleManager}.
- * This is passed to {@link #testRemoteLogSegmentLifeCycle(RemoteLogSegmentLifecycleManager)} to test
- * {@code RemoteLogMetadataCache} for several lifecycle operations.
- * <p>
- * This starts a Kafka cluster with {@link #initialize(Set, boolean)} )} with {@link #brokerCount()} no of servers. It also
- * creates the remote log metadata topic required for {@code TopicBasedRemoteLogMetadataManager}. This cluster will
- * be stopped by invoking {@link #close()}.
- */
- static class TopicBasedRemoteLogMetadataManagerWrapper extends TopicBasedRemoteLogMetadataManagerHarness implements RemoteLogSegmentLifecycleManager {
-
- private TopicIdPartition topicIdPartition;
-
- @Override
- public synchronized void initialize(TopicIdPartition topicIdPartition) {
- this.topicIdPartition = topicIdPartition;
- super.initialize(Collections.singleton(topicIdPartition), true);
- }
-
- @Override
- public void addRemoteLogSegmentMetadata(RemoteLogSegmentMetadata segmentMetadata) throws RemoteStorageException {
- try {
- // Wait until the segment is added successfully.
- remoteLogMetadataManager().addRemoteLogSegmentMetadata(segmentMetadata).get();
- } catch (Exception e) {
- throw new RemoteStorageException(e);
- }
- }
-
- @Override
- public void updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate segmentMetadataUpdate) throws RemoteStorageException {
- try {
- // Wait until the segment is updated successfully.
- remoteLogMetadataManager().updateRemoteLogSegmentMetadata(segmentMetadataUpdate).get();
- } catch (Exception e) {
- throw new RemoteStorageException(e);
- }
- }
-
- @Override
- public Optional<Long> highestOffsetForEpoch(int leaderEpoch) throws RemoteStorageException {
- return remoteLogMetadataManager().highestOffsetForEpoch(topicIdPartition, leaderEpoch);
- }
-
- @Override
- public Optional<RemoteLogSegmentMetadata> remoteLogSegmentMetadata(int leaderEpoch,
- long offset) throws RemoteStorageException {
- return remoteLogMetadataManager().remoteLogSegmentMetadata(topicIdPartition, leaderEpoch, offset);
- }
-
- @Override
- public Iterator<RemoteLogSegmentMetadata> listRemoteLogSegments(int leaderEpoch) throws RemoteStorageException {
- return remoteLogMetadataManager().listRemoteLogSegments(topicIdPartition, leaderEpoch);
- }
-
- @Override
- public Iterator<RemoteLogSegmentMetadata> listAllRemoteLogSegments() throws RemoteStorageException {
- return remoteLogMetadataManager().listRemoteLogSegments(topicIdPartition);
- }
-
- @Override
- public void close() throws IOException {
- super.close();
- }
-
- }
-
- /**
- * This is a wrapper with {@link RemoteLogMetadataCache} implementing {@link RemoteLogSegmentLifecycleManager}.
- * This is passed to {@link #testRemoteLogSegmentLifeCycle(RemoteLogSegmentLifecycleManager)} to test
- * {@code RemoteLogMetadataCache} for several lifecycle operations.
- */
- static class RemoteLogMetadataCacheWrapper implements RemoteLogSegmentLifecycleManager {
-
- private final RemoteLogMetadataCache metadataCache = new RemoteLogMetadataCache();
-
- @Override
- public void updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate segmentMetadataUpdate) throws RemoteStorageException {
- metadataCache.updateRemoteLogSegmentMetadata(segmentMetadataUpdate);
- }
-
- @Override
- public Optional<Long> highestOffsetForEpoch(int epoch) {
- return metadataCache.highestOffsetForEpoch(epoch);
- }
-
- @Override
- public Optional<RemoteLogSegmentMetadata> remoteLogSegmentMetadata(int leaderEpoch, long offset) {
- return metadataCache.remoteLogSegmentMetadata(leaderEpoch, offset);
- }
-
- @Override
- public Iterator<RemoteLogSegmentMetadata> listRemoteLogSegments(int leaderEpoch) throws RemoteResourceNotFoundException {
- return metadataCache.listRemoteLogSegments(leaderEpoch);
- }
-
- @Override
- public Iterator<RemoteLogSegmentMetadata> listAllRemoteLogSegments() {
- return metadataCache.listAllRemoteLogSegments();
- }
-
- @Override
- public void addRemoteLogSegmentMetadata(RemoteLogSegmentMetadata segmentMetadata) {
- metadataCache.addCopyInProgressSegment(segmentMetadata);
+ assertTrue(TestUtils.sameElementsWithOrder(
+ expectedSegmentsForEpoch1.iterator(), metadataManager.listRemoteLogSegments(topicIdPartition, 1)));
}
}
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
deleted file mode 100644
index 7af78e750a8..00000000000
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
+++ /dev/null
@@ -1,90 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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 org.apache.kafka.server.log.remote.metadata.storage;
-
-import kafka.api.IntegrationTestHarness;
-import kafka.utils.EmptyTestInfo;
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.common.utils.Utils;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Map;
-import java.util.Set;
-import java.util.function.Function;
-import java.util.function.Supplier;
-
-/**
- * A test harness class that brings up 3 brokers and registers {@link TopicBasedRemoteLogMetadataManager} on broker with id as 0.
- */
-public class TopicBasedRemoteLogMetadataManagerHarness extends IntegrationTestHarness {
-
- private TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager;
-
- protected Map<String, Object> overrideRemoteLogMetadataManagerProps() {
- return Collections.emptyMap();
- }
-
- public void initialize(Set<TopicIdPartition> topicIdPartitions,
- boolean startConsumerThread) {
- initialize(topicIdPartitions, startConsumerThread, RemotePartitionMetadataStore::new);
- }
-
- public void initialize(Set<TopicIdPartition> topicIdPartitions,
- boolean startConsumerThread,
- Supplier<RemotePartitionMetadataStore> remotePartitionMetadataStoreSupplier) {
- // Call setup to start the cluster.
- super.setUp(new EmptyTestInfo());
-
- initializeRemoteLogMetadataManager(topicIdPartitions, startConsumerThread, RemoteLogMetadataTopicPartitioner::new, remotePartitionMetadataStoreSupplier);
- }
-
- public void initializeRemoteLogMetadataManager(Set<TopicIdPartition> topicIdPartitions,
- boolean startConsumerThread,
- Function<Integer, RemoteLogMetadataTopicPartitioner> remoteLogMetadataTopicPartitioner,
- Supplier<RemotePartitionMetadataStore> remotePartitionMetadataStoreSupplier) {
-
- topicBasedRemoteLogMetadataManager = RemoteLogMetadataManagerTestUtils.builder()
- .topicIdPartitions(topicIdPartitions)
- .bootstrapServers(bootstrapServers(listenerName()))
- .startConsumerThread(startConsumerThread)
- .remoteLogMetadataTopicPartitioner(remoteLogMetadataTopicPartitioner)
- .remotePartitionMetadataStore(remotePartitionMetadataStoreSupplier)
- .overrideRemoteLogMetadataManagerProps(overrideRemoteLogMetadataManagerProps())
- .build();
- }
-
- @Override
- public int brokerCount() {
- return 3;
- }
-
- protected TopicBasedRemoteLogMetadataManager remoteLogMetadataManager() {
- return topicBasedRemoteLogMetadataManager;
- }
-
- public void close() throws IOException {
- closeRemoteLogMetadataManager();
-
- // Stop the servers and zookeeper.
- tearDown();
- }
-
- public void closeRemoteLogMetadataManager() {
- Utils.closeQuietly(topicBasedRemoteLogMetadataManager, "TopicBasedRemoteLogMetadataManager");
- }
-}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
index 84b98dcb5be..dee686dc036 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
@@ -54,7 +54,6 @@ public class TopicBasedRemoteLogMetadataManagerRestartTest {
private TopicBasedRemoteLogMetadataManager createTopicBasedRemoteLogMetadataManager() {
return RemoteLogMetadataManagerTestUtils.builder()
- .topicIdPartitions(Collections.emptySet())
.bootstrapServers(clusterInstance.bootstrapServers())
.startConsumerThread(true)
.remoteLogMetadataTopicPartitioner(RemoteLogMetadataTopicPartitioner::new)
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerWrapperWithHarness.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerWrapperWithHarness.java
deleted file mode 100644
index 70ce9190d81..00000000000
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerWrapperWithHarness.java
+++ /dev/null
@@ -1,105 +0,0 @@
-/*
- * Licensed to the Apache Software Foundation (ASF) under one or more
- * contributor license agreements. See the NOTICE file distributed with
- * this work for additional information regarding copyright ownership.
- * The ASF licenses this file to You 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 org.apache.kafka.server.log.remote.metadata.storage;
-
-import org.apache.kafka.common.TopicIdPartition;
-import org.apache.kafka.server.log.remote.storage.RemoteLogMetadataManager;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
-import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadataUpdate;
-import org.apache.kafka.server.log.remote.storage.RemotePartitionDeleteMetadata;
-import org.apache.kafka.server.log.remote.storage.RemoteStorageException;
-
-import java.io.IOException;
-import java.util.Collections;
-import java.util.Iterator;
-import java.util.Map;
-import java.util.Optional;
-import java.util.Set;
-import java.util.concurrent.CompletableFuture;
-
-public class TopicBasedRemoteLogMetadataManagerWrapperWithHarness implements RemoteLogMetadataManager {
-
- private final TopicBasedRemoteLogMetadataManagerHarness remoteLogMetadataManagerHarness = new TopicBasedRemoteLogMetadataManagerHarness();
-
- @Override
- public CompletableFuture<Void> addRemoteLogSegmentMetadata(RemoteLogSegmentMetadata remoteLogSegmentMetadata) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().addRemoteLogSegmentMetadata(remoteLogSegmentMetadata);
- }
-
- @Override
- public CompletableFuture<Void> updateRemoteLogSegmentMetadata(RemoteLogSegmentMetadataUpdate remoteLogSegmentMetadataUpdate) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().updateRemoteLogSegmentMetadata(remoteLogSegmentMetadataUpdate);
- }
-
- @Override
- public Optional<RemoteLogSegmentMetadata> remoteLogSegmentMetadata(TopicIdPartition topicIdPartition,
- int epochForOffset,
- long offset) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().remoteLogSegmentMetadata(topicIdPartition, epochForOffset, offset);
- }
-
- @Override
- public Optional<Long> highestOffsetForEpoch(TopicIdPartition topicIdPartition,
- int leaderEpoch) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().highestOffsetForEpoch(topicIdPartition, leaderEpoch);
- }
-
- @Override
- public CompletableFuture<Void> putRemotePartitionDeleteMetadata(RemotePartitionDeleteMetadata remotePartitionDeleteMetadata) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().putRemotePartitionDeleteMetadata(remotePartitionDeleteMetadata);
- }
-
- @Override
- public Iterator<RemoteLogSegmentMetadata> listRemoteLogSegments(TopicIdPartition topicIdPartition) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().listRemoteLogSegments(topicIdPartition);
- }
-
- @Override
- public Iterator<RemoteLogSegmentMetadata> listRemoteLogSegments(TopicIdPartition topicIdPartition,
- int leaderEpoch) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().listRemoteLogSegments(topicIdPartition, leaderEpoch);
- }
-
- @Override
- public void onPartitionLeadershipChanges(Set<TopicIdPartition> leaderPartitions,
- Set<TopicIdPartition> followerPartitions) {
-
- remoteLogMetadataManagerHarness.remoteLogMetadataManager().onPartitionLeadershipChanges(leaderPartitions, followerPartitions);
- }
-
- @Override
- public void onStopPartitions(Set<TopicIdPartition> partitions) {
- remoteLogMetadataManagerHarness.remoteLogMetadataManager().onStopPartitions(partitions);
- }
-
- @Override
- public long remoteLogSize(TopicIdPartition topicIdPartition, int leaderEpoch) throws RemoteStorageException {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager().remoteLogSize(topicIdPartition, leaderEpoch);
- }
-
- @Override
- public void close() throws IOException {
- remoteLogMetadataManagerHarness.close();
- }
-
- @Override
- public void configure(Map<String, ?> configs) {
- // This will make sure the cluster is up and TopicBasedRemoteLogMetadataManager is initialized.
- remoteLogMetadataManagerHarness.initialize(Collections.emptySet(), true);
- remoteLogMetadataManagerHarness.remoteLogMetadataManager().configure(configs);
- }
-}
(kafka) 01/12: KAFKA-16866 Used the right constant in RemoteLogManagerTest#testFetchQuotaManagerConfig (#16152)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 68d92a5b43f001a06e4c5d9c6f244c6cc33595b1
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Fri May 31 22:44:31 2024 +0530
KAFKA-16866 Used the right constant in RemoteLogManagerTest#testFetchQuotaManagerConfig (#16152)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java | 9 ++++++---
1 file changed, 6 insertions(+), 3 deletions(-)
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 72154c9fc22..0ba5d63a8da 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -128,6 +128,9 @@ import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemo
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND;
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM;
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM;
+import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS;
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_LOG_METADATA_MANAGER_CONFIG_PREFIX;
import static org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig.DEFAULT_REMOTE_STORAGE_MANAGER_CONFIG_PREFIX;
import static org.apache.kafka.server.log.remote.storage.RemoteStorageMetrics.REMOTE_LOG_MANAGER_TASKS_AVG_IDLE_PERCENT_METRIC;
@@ -2701,9 +2704,9 @@ public class RemoteLogManagerTest {
Properties defaultProps = new Properties();
RemoteLogManagerConfig defaultRlmConfig = createRLMConfig(defaultProps);
RLMQuotaManagerConfig defaultConfig = RemoteLogManager.fetchQuotaManagerConfig(defaultRlmConfig);
- assertEquals(DEFAULT_REMOTE_LOG_MANAGER_COPY_MAX_BYTES_PER_SECOND, defaultConfig.quotaBytesPerSecond());
- assertEquals(DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_NUM, defaultConfig.numQuotaSamples());
- assertEquals(DEFAULT_REMOTE_LOG_MANAGER_COPY_QUOTA_WINDOW_SIZE_SECONDS, defaultConfig.quotaWindowSizeSeconds());
+ assertEquals(DEFAULT_REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND, defaultConfig.quotaBytesPerSecond());
+ assertEquals(DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_NUM, defaultConfig.numQuotaSamples());
+ assertEquals(DEFAULT_REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS, defaultConfig.quotaWindowSizeSeconds());
Properties customProps = new Properties();
customProps.put(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_FETCH_MAX_BYTES_PER_SECOND_PROP, 100);
(kafka) 04/12: KAFKA-16859 Cleanup check if tiered storage is enabled (#16153)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit f9c37032ffc316de86fee3d8096a566663a56451
Author: Ken Huang <10...@users.noreply.github.com>
AuthorDate: Mon Jun 3 12:04:58 2024 +0900
KAFKA-16859 Cleanup check if tiered storage is enabled (#16153)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../kafka/server/builders/KafkaApisBuilder.java | 2 +-
.../server/builders/ReplicaManagerBuilder.java | 2 +-
core/src/main/scala/kafka/log/LogManager.scala | 2 +-
.../src/main/scala/kafka/server/BrokerServer.scala | 2 +-
.../main/scala/kafka/server/ConfigHandler.scala | 2 +-
.../server/ControllerConfigurationValidator.scala | 3 ++-
core/src/main/scala/kafka/server/KafkaConfig.scala | 4 +---
core/src/main/scala/kafka/server/KafkaServer.scala | 2 +-
core/src/main/scala/kafka/zk/AdminZkClient.scala | 4 ++--
.../kafka/log/remote/RemoteLogManagerTest.java | 2 +-
.../test/scala/unit/kafka/log/LogConfigTest.scala | 26 +++++++++++-----------
.../unit/kafka/server/ReplicaManagerTest.scala | 4 ++--
12 files changed, 27 insertions(+), 28 deletions(-)
diff --git a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
index 6ffd741f4fc..1d422461678 100644
--- a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
+++ b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
@@ -179,7 +179,7 @@ public class KafkaApisBuilder {
if (metrics == null) throw new RuntimeException("You must set metrics");
if (quotas == null) throw new RuntimeException("You must set quotas");
if (fetchManager == null) throw new RuntimeException("You must set fetchManager");
- if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled());
+ if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig().enableRemoteStorageSystem());
if (apiVersionManager == null) throw new RuntimeException("You must set apiVersionManager");
return new KafkaApis(requestChannel,
diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
index 82aa75909ab..5e8cf2dcdc6 100644
--- a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
@@ -185,7 +185,7 @@ public class ReplicaManagerBuilder {
if (metadataCache == null) throw new RuntimeException("You must set metadataCache");
if (logDirFailureChannel == null) throw new RuntimeException("You must set logDirFailureChannel");
if (alterPartitionManager == null) throw new RuntimeException("You must set alterIsrManager");
- if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled());
+ if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig().enableRemoteStorageSystem());
// Initialize metrics in the end just before passing it to ReplicaManager to ensure ReplicaManager closes the
// metrics correctly. There might be a resource leak if it is initialized and an exception occurs between
// its initialization and creation of ReplicaManager.
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index 8bc6db2dff6..45d6ab7908d 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -1562,7 +1562,7 @@ object LogManager {
keepPartitionMetadataFile: Boolean): LogManager = {
val defaultProps = config.extractLogConfigMap
- LogConfig.validateBrokerLogConfigValues(defaultProps, config.isRemoteLogStorageSystemEnabled)
+ LogConfig.validateBrokerLogConfigValues(defaultProps, config.remoteLogManagerConfig.enableRemoteStorageSystem())
val defaultLogConfig = new LogConfig(defaultProps)
val cleanerConfig = LogCleaner.cleanerConfig(config)
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index 94bcf321420..baafc3e658f 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -186,7 +186,7 @@ class BrokerServer(
kafkaScheduler.startup()
/* register broker metrics */
- brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled)
+ brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.enableRemoteStorageSystem())
quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-")
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index 1d5702e76e4..ed9260b2194 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -70,7 +70,7 @@ class TopicConfigHandler(private val replicaManager: ReplicaManager,
val logs = logManager.logsByTopic(topic)
val wasRemoteLogEnabledBeforeUpdate = logs.exists(_.remoteLogEnabled())
- logManager.updateTopicConfig(topic, props, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ logManager.updateTopicConfig(topic, props, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
maybeBootstrapRemoteLogComponents(topic, logs, wasRemoteLogEnabledBeforeUpdate)
}
diff --git a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
index 15eb1eff04a..f957b65ddd1 100644
--- a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
+++ b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
@@ -107,7 +107,8 @@ class ControllerConfigurationValidator(kafkaConfig: KafkaConfig) extends Configu
throw new InvalidConfigurationException("Null value not supported for topic configs: " +
nullTopicConfigs.mkString(","))
}
- LogConfig.validate(properties, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ LogConfig.validate(properties, kafkaConfig.extractLogConfigMap,
+ kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
case BROKER => validateBrokerName(resource.name())
case CLIENT_METRICS =>
val properties = new Properties()
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index b812f299ecb..3635d5a3edd 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -17,7 +17,7 @@
package kafka.server
-import java.{lang, util}
+import java.util
import java.util.concurrent.TimeUnit
import java.util.{Collections, Properties}
import kafka.cluster.EndPoint
@@ -1207,8 +1207,6 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def usesTopicId: Boolean =
usesSelfManagedQuorum || interBrokerProtocolVersion.isTopicIdsSupported
-
- val isRemoteLogStorageSystemEnabled: lang.Boolean = getBoolean(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP)
def logLocalRetentionBytes: java.lang.Long = getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_BYTES_PROP)
def logLocalRetentionMs: java.lang.Long = getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP)
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index b913b72fc9c..9c807c79da5 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -276,7 +276,7 @@ class KafkaServer(
createCurrentControllerIdMetric()
/* register broker metrics */
- _brokerTopicStats = new BrokerTopicStats(config.isRemoteLogStorageSystemEnabled)
+ _brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.enableRemoteStorageSystem())
quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse(""))
KafkaBroker.notifyClusterListeners(clusterId, kafkaMetricsReporters ++ metrics.reporters.asScala)
diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala
index efecfe854bb..604e03c7ed4 100644
--- a/core/src/main/scala/kafka/zk/AdminZkClient.scala
+++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala
@@ -163,7 +163,7 @@ class AdminZkClient(zkClient: KafkaZkClient,
LogConfig.validate(config,
kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()),
- kafkaConfig.exists(_.isRemoteLogStorageSystemEnabled))
+ kafkaConfig.exists(_.remoteLogManagerConfig.enableRemoteStorageSystem()))
}
private def writeTopicPartitionAssignment(topic: String, replicaAssignment: Map[Int, ReplicaAssignment],
@@ -481,7 +481,7 @@ class AdminZkClient(zkClient: KafkaZkClient,
// remove the topic overrides
LogConfig.validate(configs,
kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()),
- kafkaConfig.exists(_.isRemoteLogStorageSystemEnabled))
+ kafkaConfig.exists(_.remoteLogManagerConfig.enableRemoteStorageSystem()))
}
/**
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 0ba5d63a8da..50b581fdf4e 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -222,7 +222,7 @@ public class RemoteLogManagerTest {
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true");
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP, "100");
remoteLogManagerConfig = createRLMConfig(props);
- brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).isRemoteLogStorageSystemEnabled());
+ brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig().enableRemoteStorageSystem());
remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time,
tp -> Optional.of(mockLog),
diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
index 7be9bcd0267..2bf147b528b 100644
--- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
@@ -298,7 +298,7 @@ class LogConfigTest {
props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs.toString)
props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes.toString)
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
}
@Test
@@ -310,17 +310,17 @@ class LogConfigTest {
val logProps = new Properties()
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact")
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete")
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
}
@ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}")
@@ -333,10 +333,10 @@ class LogConfigTest {
val logProps = new Properties()
logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
if (sysRemoteStorageEnabled) {
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
} else {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
assertTrue(message.getMessage.contains("Tiered Storage functionality is disabled in the broker"))
}
}
@@ -356,10 +356,10 @@ class LogConfigTest {
logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500")
if (sysRemoteStorageEnabled) {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG))
} else {
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
}
}
@@ -378,10 +378,10 @@ class LogConfigTest {
logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128")
if (sysRemoteStorageEnabled) {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG))
} else {
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
}
}
@@ -396,10 +396,10 @@ class LogConfigTest {
if (sysRemoteStorageEnabled) {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled))
+ () => LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG))
} else {
- LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.isRemoteLogStorageSystemEnabled)
+ LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
}
}
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 9f820155c8e..65cbcc7bd70 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -4095,7 +4095,7 @@ class ReplicaManagerTest {
val config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props)
val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
val mockLog = mock(classOf[UnifiedLog])
- val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).isRemoteLogStorageSystemEnabled)
+ val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem())
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
@@ -4195,7 +4195,7 @@ class ReplicaManagerTest {
val config = new AbstractConfig(RemoteLogManagerConfig.CONFIG_DEF, props)
val remoteLogManagerConfig = new RemoteLogManagerConfig(config)
val dummyLog = mock(classOf[UnifiedLog])
- val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).isRemoteLogStorageSystemEnabled)
+ val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem())
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
(kafka) 03/12: MINOR: Fix missing wait topic finished in TopicBasedRemoteLogMetadataManagerRestartTest (#16171)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 2273e061389c42686b12317f625523b46fd27a2b
Author: Kuan-Po (Cooper) Tseng <br...@gmail.com>
AuthorDate: Mon Jun 3 03:59:56 2024 +0800
MINOR: Fix missing wait topic finished in TopicBasedRemoteLogMetadataManagerRestartTest (#16171)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../storage/TopicBasedRemoteLogMetadataManagerRestartTest.java | 6 ++++--
1 file changed, 4 insertions(+), 2 deletions(-)
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
index 07b08145731..84b98dcb5be 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
@@ -72,8 +72,10 @@ public class TopicBasedRemoteLogMetadataManagerRestartTest {
NewTopic newLeaderTopic = new NewTopic(leaderTopic, Collections.singletonMap(0, Arrays.asList(0, 1, 2)));
// Set broker id 1 as the first entry which is taken as the leader.
NewTopic newFollowerTopic = new NewTopic(followerTopic, Collections.singletonMap(0, Arrays.asList(1, 2, 0)));
- admin.createTopics(Arrays.asList(newLeaderTopic, newFollowerTopic));
+ admin.createTopics(Arrays.asList(newLeaderTopic, newFollowerTopic)).all().get();
}
+ clusterInstance.waitForTopic(leaderTopic, 1);
+ clusterInstance.waitForTopic(followerTopic, 1);
final TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
final TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));
@@ -122,4 +124,4 @@ public class TopicBasedRemoteLogMetadataManagerRestartTest {
topicBasedRemoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition)));
}
}
-}
\ No newline at end of file
+}
(kafka) 11/12: KAFKA-16885 Renamed the enableRemoteStorageSystem to isRemoteStorageSystemEnabled (#16256)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit 781b93b00d99a63dee0ed6650a2d398f7791cea3
Author: Chia Chuan Yu <yu...@gmail.com>
AuthorDate: Mon Jun 10 02:14:15 2024 +0800
KAFKA-16885 Renamed the enableRemoteStorageSystem to isRemoteStorageSystemEnabled (#16256)
Reviewers: Kamal Chandraprakash <kc...@uber.com>, Chia-Ping Tsai <ch...@gmail.com>
---
.../java/kafka/log/remote/RemoteLogManager.java | 2 +-
.../kafka/server/builders/KafkaApisBuilder.java | 2 +-
.../server/builders/ReplicaManagerBuilder.java | 2 +-
core/src/main/scala/kafka/log/LogManager.scala | 4 ++--
.../src/main/scala/kafka/server/BrokerServer.scala | 4 ++--
.../main/scala/kafka/server/ConfigHandler.scala | 2 +-
.../server/ControllerConfigurationValidator.scala | 2 +-
core/src/main/scala/kafka/server/KafkaServer.scala | 4 ++--
core/src/main/scala/kafka/zk/AdminZkClient.scala | 4 ++--
.../kafka/log/remote/RemoteLogManagerTest.java | 2 +-
.../test/scala/unit/kafka/log/LogConfigTest.scala | 26 +++++++++++-----------
.../test/scala/unit/kafka/log/LogLoaderTest.scala | 2 +-
.../unit/kafka/server/ReplicaManagerTest.scala | 6 ++---
.../log/remote/storage/RemoteLogManagerConfig.java | 2 +-
14 files changed, 32 insertions(+), 32 deletions(-)
diff --git a/core/src/main/java/kafka/log/remote/RemoteLogManager.java b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
index 5b0d91ff439..de5fbd85f16 100644
--- a/core/src/main/java/kafka/log/remote/RemoteLogManager.java
+++ b/core/src/main/java/kafka/log/remote/RemoteLogManager.java
@@ -378,7 +378,7 @@ public class RemoteLogManager implements Closeable {
Map<String, Uuid> topicIds) {
LOGGER.debug("Received leadership changes for leaders: {} and followers: {}", partitionsBecomeLeader, partitionsBecomeFollower);
- if (this.rlmConfig.enableRemoteStorageSystem() && !isRemoteLogManagerConfigured()) {
+ if (this.rlmConfig.isRemoteStorageSystemEnabled() && !isRemoteLogManagerConfigured()) {
throw new KafkaException("RemoteLogManager is not configured when remote storage system is enabled");
}
diff --git a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
index 1d422461678..56607723604 100644
--- a/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
+++ b/core/src/main/java/kafka/server/builders/KafkaApisBuilder.java
@@ -179,7 +179,7 @@ public class KafkaApisBuilder {
if (metrics == null) throw new RuntimeException("You must set metrics");
if (quotas == null) throw new RuntimeException("You must set quotas");
if (fetchManager == null) throw new RuntimeException("You must set fetchManager");
- if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig().enableRemoteStorageSystem());
+ if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
if (apiVersionManager == null) throw new RuntimeException("You must set apiVersionManager");
return new KafkaApis(requestChannel,
diff --git a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
index 5e8cf2dcdc6..7cac33200d2 100644
--- a/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
+++ b/core/src/main/java/kafka/server/builders/ReplicaManagerBuilder.java
@@ -185,7 +185,7 @@ public class ReplicaManagerBuilder {
if (metadataCache == null) throw new RuntimeException("You must set metadataCache");
if (logDirFailureChannel == null) throw new RuntimeException("You must set logDirFailureChannel");
if (alterPartitionManager == null) throw new RuntimeException("You must set alterIsrManager");
- if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig().enableRemoteStorageSystem());
+ if (brokerTopicStats == null) brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig().isRemoteStorageSystemEnabled());
// Initialize metrics in the end just before passing it to ReplicaManager to ensure ReplicaManager closes the
// metrics correctly. There might be a resource leak if it is initialized and an exception occurs between
// its initialization and creation of ReplicaManager.
diff --git a/core/src/main/scala/kafka/log/LogManager.scala b/core/src/main/scala/kafka/log/LogManager.scala
index 45d6ab7908d..8908aadf459 100755
--- a/core/src/main/scala/kafka/log/LogManager.scala
+++ b/core/src/main/scala/kafka/log/LogManager.scala
@@ -1562,7 +1562,7 @@ object LogManager {
keepPartitionMetadataFile: Boolean): LogManager = {
val defaultProps = config.extractLogConfigMap
- LogConfig.validateBrokerLogConfigValues(defaultProps, config.remoteLogManagerConfig.enableRemoteStorageSystem())
+ LogConfig.validateBrokerLogConfigValues(defaultProps, config.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
val defaultLogConfig = new LogConfig(defaultProps)
val cleanerConfig = LogCleaner.cleanerConfig(config)
@@ -1586,7 +1586,7 @@ object LogManager {
time = time,
keepPartitionMetadataFile = keepPartitionMetadataFile,
interBrokerProtocolVersion = config.interBrokerProtocolVersion,
- remoteStorageSystemEnable = config.remoteLogManagerConfig.enableRemoteStorageSystem(),
+ remoteStorageSystemEnable = config.remoteLogManagerConfig.isRemoteStorageSystemEnabled(),
initialTaskDelayMs = config.logInitialTaskDelayMs)
}
diff --git a/core/src/main/scala/kafka/server/BrokerServer.scala b/core/src/main/scala/kafka/server/BrokerServer.scala
index baafc3e658f..e143dd26668 100644
--- a/core/src/main/scala/kafka/server/BrokerServer.scala
+++ b/core/src/main/scala/kafka/server/BrokerServer.scala
@@ -186,7 +186,7 @@ class BrokerServer(
kafkaScheduler.startup()
/* register broker metrics */
- brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.enableRemoteStorageSystem())
+ brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
quotaManagers = QuotaFactory.instantiate(config, metrics, time, s"broker-${config.nodeId}-")
@@ -621,7 +621,7 @@ class BrokerServer(
}
protected def createRemoteLogManager(): Option[RemoteLogManager] = {
- if (config.remoteLogManagerConfig.enableRemoteStorageSystem()) {
+ if (config.remoteLogManagerConfig.isRemoteStorageSystemEnabled()) {
Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
(tp: TopicPartition) => logManager.getLog(tp).asJava,
(tp: TopicPartition, remoteLogStartOffset: java.lang.Long) => {
diff --git a/core/src/main/scala/kafka/server/ConfigHandler.scala b/core/src/main/scala/kafka/server/ConfigHandler.scala
index ed9260b2194..e441b30a3bd 100644
--- a/core/src/main/scala/kafka/server/ConfigHandler.scala
+++ b/core/src/main/scala/kafka/server/ConfigHandler.scala
@@ -70,7 +70,7 @@ class TopicConfigHandler(private val replicaManager: ReplicaManager,
val logs = logManager.logsByTopic(topic)
val wasRemoteLogEnabledBeforeUpdate = logs.exists(_.remoteLogEnabled())
- logManager.updateTopicConfig(topic, props, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ logManager.updateTopicConfig(topic, props, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
maybeBootstrapRemoteLogComponents(topic, logs, wasRemoteLogEnabledBeforeUpdate)
}
diff --git a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
index f957b65ddd1..b99065b573e 100644
--- a/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
+++ b/core/src/main/scala/kafka/server/ControllerConfigurationValidator.scala
@@ -108,7 +108,7 @@ class ControllerConfigurationValidator(kafkaConfig: KafkaConfig) extends Configu
nullTopicConfigs.mkString(","))
}
LogConfig.validate(properties, kafkaConfig.extractLogConfigMap,
- kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
case BROKER => validateBrokerName(resource.name())
case CLIENT_METRICS =>
val properties = new Properties()
diff --git a/core/src/main/scala/kafka/server/KafkaServer.scala b/core/src/main/scala/kafka/server/KafkaServer.scala
index 9c807c79da5..5b6e04e5a0e 100755
--- a/core/src/main/scala/kafka/server/KafkaServer.scala
+++ b/core/src/main/scala/kafka/server/KafkaServer.scala
@@ -276,7 +276,7 @@ class KafkaServer(
createCurrentControllerIdMetric()
/* register broker metrics */
- _brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.enableRemoteStorageSystem())
+ _brokerTopicStats = new BrokerTopicStats(config.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
quotaManagers = QuotaFactory.instantiate(config, metrics, time, threadNamePrefix.getOrElse(""))
KafkaBroker.notifyClusterListeners(clusterId, kafkaMetricsReporters ++ metrics.reporters.asScala)
@@ -690,7 +690,7 @@ class KafkaServer(
}
protected def createRemoteLogManager(): Option[RemoteLogManager] = {
- if (config.remoteLogManagerConfig.enableRemoteStorageSystem()) {
+ if (config.remoteLogManagerConfig.isRemoteStorageSystemEnabled()) {
Some(new RemoteLogManager(config.remoteLogManagerConfig, config.brokerId, config.logDirs.head, clusterId, time,
(tp: TopicPartition) => logManager.getLog(tp).asJava,
(tp: TopicPartition, remoteLogStartOffset: java.lang.Long) => {
diff --git a/core/src/main/scala/kafka/zk/AdminZkClient.scala b/core/src/main/scala/kafka/zk/AdminZkClient.scala
index 604e03c7ed4..cd9153c07dc 100644
--- a/core/src/main/scala/kafka/zk/AdminZkClient.scala
+++ b/core/src/main/scala/kafka/zk/AdminZkClient.scala
@@ -163,7 +163,7 @@ class AdminZkClient(zkClient: KafkaZkClient,
LogConfig.validate(config,
kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()),
- kafkaConfig.exists(_.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ kafkaConfig.exists(_.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
}
private def writeTopicPartitionAssignment(topic: String, replicaAssignment: Map[Int, ReplicaAssignment],
@@ -481,7 +481,7 @@ class AdminZkClient(zkClient: KafkaZkClient,
// remove the topic overrides
LogConfig.validate(configs,
kafkaConfig.map(_.extractLogConfigMap).getOrElse(Collections.emptyMap()),
- kafkaConfig.exists(_.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ kafkaConfig.exists(_.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
}
/**
diff --git a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
index 9badaabf419..edc85cdf28b 100644
--- a/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
+++ b/core/src/test/java/kafka/log/remote/RemoteLogManagerTest.java
@@ -221,7 +221,7 @@ public class RemoteLogManagerTest {
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP, "true");
props.setProperty(RemoteLogManagerConfig.REMOTE_LOG_MANAGER_TASK_INTERVAL_MS_PROP, "100");
remoteLogManagerConfig = createRLMConfig(props);
- brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig().enableRemoteStorageSystem());
+ brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig().isRemoteStorageSystemEnabled());
remoteLogManager = new RemoteLogManager(remoteLogManagerConfig, brokerId, logDir, clusterId, time,
tp -> Optional.of(mockLog),
diff --git a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
index 2bf147b528b..bd1ee39a483 100644
--- a/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogConfigTest.scala
@@ -298,7 +298,7 @@ class LogConfigTest {
props.put(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG, localRetentionMs.toString)
props.put(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG, localRetentionBytes.toString)
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(props, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
}
@Test
@@ -310,17 +310,17 @@ class LogConfigTest {
val logProps = new Properties()
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_DELETE)
logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, TopicConfig.CLEANUP_POLICY_COMPACT)
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "delete,compact")
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
logProps.put(TopicConfig.CLEANUP_POLICY_CONFIG, "compact,delete")
assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
}
@ParameterizedTest(name = "testEnableRemoteLogStorage with sysRemoteStorageEnabled: {0}")
@@ -333,10 +333,10 @@ class LogConfigTest {
val logProps = new Properties()
logProps.put(TopicConfig.REMOTE_LOG_STORAGE_ENABLE_CONFIG, "true")
if (sysRemoteStorageEnabled) {
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
} else {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
assertTrue(message.getMessage.contains("Tiered Storage functionality is disabled in the broker"))
}
}
@@ -356,10 +356,10 @@ class LogConfigTest {
logProps.put(TopicConfig.RETENTION_MS_CONFIG, "500")
if (sysRemoteStorageEnabled) {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_MS_CONFIG))
} else {
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
}
}
@@ -378,10 +378,10 @@ class LogConfigTest {
logProps.put(TopicConfig.RETENTION_BYTES_CONFIG, "128")
if (sysRemoteStorageEnabled) {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG))
} else {
- LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ LogConfig.validate(logProps, kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
}
}
@@ -396,10 +396,10 @@ class LogConfigTest {
if (sysRemoteStorageEnabled) {
val message = assertThrows(classOf[ConfigException],
- () => LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem()))
+ () => LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled()))
assertTrue(message.getMessage.contains(TopicConfig.LOCAL_LOG_RETENTION_BYTES_CONFIG))
} else {
- LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.enableRemoteStorageSystem())
+ LogConfig.validateBrokerLogConfigValues(kafkaConfig.extractLogConfigMap, kafkaConfig.remoteLogManagerConfig.isRemoteStorageSystemEnabled())
}
}
diff --git a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
index 1a781a93ea6..53e4229a276 100644
--- a/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
+++ b/core/src/test/scala/unit/kafka/log/LogLoaderTest.scala
@@ -128,7 +128,7 @@ class LogLoaderTest {
logDirFailureChannel = logDirFailureChannel,
time = time,
keepPartitionMetadataFile = config.usesTopicId,
- remoteStorageSystemEnable = config.remoteLogManagerConfig.enableRemoteStorageSystem(),
+ remoteStorageSystemEnable = config.remoteLogManagerConfig.isRemoteStorageSystemEnabled(),
initialTaskDelayMs = config.logInitialTaskDelayMs) {
override def loadLog(logDir: File, hadCleanShutdown: Boolean, recoveryPoints: Map[TopicPartition, Long],
diff --git a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
index 313b679c605..a6d0d1f3293 100644
--- a/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
+++ b/core/src/test/scala/unit/kafka/server/ReplicaManagerTest.scala
@@ -3389,7 +3389,7 @@ class ReplicaManagerTest {
when(mockLog.remoteLogEnabled()).thenReturn(enableRemoteStorage)
when(mockLog.remoteStorageSystemEnable).thenReturn(enableRemoteStorage)
val aliveBrokers = aliveBrokerIds.map(brokerId => new Node(brokerId, s"host$brokerId", brokerId))
- brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem)
+ brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.isRemoteStorageSystemEnabled)
val metadataCache: MetadataCache = mock(classOf[MetadataCache])
when(metadataCache.topicIdInfo()).thenReturn((topicIds.asJava, topicNames.asJava))
@@ -4094,7 +4094,7 @@ class ReplicaManagerTest {
props.put(RemoteLogManagerConfig.REMOTE_LOG_READER_THREADS_PROP, 2.toString)
val remoteLogManagerConfig = new RemoteLogManagerConfig(props)
val mockLog = mock(classOf[UnifiedLog])
- val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem())
+ val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.isRemoteStorageSystemEnabled())
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
@@ -4193,7 +4193,7 @@ class ReplicaManagerTest {
props.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_CLASS_NAME_PROP, classOf[NoOpRemoteLogMetadataManager].getName)
val remoteLogManagerConfig = new RemoteLogManagerConfig(props)
val dummyLog = mock(classOf[UnifiedLog])
- val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.enableRemoteStorageSystem())
+ val brokerTopicStats = new BrokerTopicStats(KafkaConfig.fromProps(props).remoteLogManagerConfig.isRemoteStorageSystemEnabled())
val remoteLogManager = new RemoteLogManager(
remoteLogManagerConfig,
0,
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index 8224ab23825..6323ba65bfe 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -358,7 +358,7 @@ public final class RemoteLogManagerConfig extends AbstractConfig {
super(configDef(), props);
}
- public boolean enableRemoteStorageSystem() {
+ public boolean isRemoteStorageSystemEnabled() {
return getBoolean(REMOTE_LOG_STORAGE_SYSTEM_ENABLE_PROP);
}
(kafka) 02/12: KAFKA-16785 Migrate TopicBasedRemoteLogMetadataManagerRestartTest to new test infra (#16170)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit c6f0db3c6069fd49564519b1bf26c10e3cd237e6
Author: Kuan-Po (Cooper) Tseng <br...@gmail.com>
AuthorDate: Sun Jun 2 22:18:53 2024 +0800
KAFKA-16785 Migrate TopicBasedRemoteLogMetadataManagerRestartTest to new test infra (#16170)
Reviewers: Chia-Ping Tsai <ch...@gmail.com>
---
.../TopicBasedRemoteLogMetadataManagerHarness.java | 7 +-
...icBasedRemoteLogMetadataManagerRestartTest.java | 159 ++++++++-------------
2 files changed, 62 insertions(+), 104 deletions(-)
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
index a063fa8820a..7af78e750a8 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerHarness.java
@@ -53,12 +53,6 @@ public class TopicBasedRemoteLogMetadataManagerHarness extends IntegrationTestHa
initializeRemoteLogMetadataManager(topicIdPartitions, startConsumerThread, RemoteLogMetadataTopicPartitioner::new, remotePartitionMetadataStoreSupplier);
}
- public void initializeRemoteLogMetadataManager(Set<TopicIdPartition> topicIdPartitions,
- boolean startConsumerThread,
- Function<Integer, RemoteLogMetadataTopicPartitioner> remoteLogMetadataTopicPartitioner) {
- initializeRemoteLogMetadataManager(topicIdPartitions, startConsumerThread, remoteLogMetadataTopicPartitioner, RemotePartitionMetadataStore::new);
- }
-
public void initializeRemoteLogMetadataManager(Set<TopicIdPartition> topicIdPartitions,
boolean startConsumerThread,
Function<Integer, RemoteLogMetadataTopicPartitioner> remoteLogMetadataTopicPartitioner,
@@ -70,6 +64,7 @@ public class TopicBasedRemoteLogMetadataManagerHarness extends IntegrationTestHa
.startConsumerThread(startConsumerThread)
.remoteLogMetadataTopicPartitioner(remoteLogMetadataTopicPartitioner)
.remotePartitionMetadataStore(remotePartitionMetadataStoreSupplier)
+ .overrideRemoteLogMetadataManagerProps(overrideRemoteLogMetadataManagerProps())
.build();
}
diff --git a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
index c599259ed94..07b08145731 100644
--- a/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
+++ b/storage/src/test/java/org/apache/kafka/server/log/remote/metadata/storage/TopicBasedRemoteLogMetadataManagerRestartTest.java
@@ -16,6 +16,11 @@
*/
package org.apache.kafka.server.log.remote.metadata.storage;
+import kafka.test.ClusterInstance;
+import kafka.test.annotation.ClusterTest;
+import kafka.test.junit.ClusterTestExtensions;
+import org.apache.kafka.clients.admin.Admin;
+import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.common.TopicIdPartition;
import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.Uuid;
@@ -24,139 +29,97 @@ import org.apache.kafka.common.utils.Time;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentId;
import org.apache.kafka.server.log.remote.storage.RemoteLogSegmentMetadata;
import org.apache.kafka.test.TestUtils;
-import org.junit.jupiter.api.AfterEach;
import org.junit.jupiter.api.Assertions;
-import org.junit.jupiter.api.BeforeEach;
-import org.junit.jupiter.api.Test;
-import scala.collection.JavaConverters;
-import scala.collection.Seq;
+import org.junit.jupiter.api.Tag;
+import org.junit.jupiter.api.extension.ExtendWith;
-import java.io.IOException;
-import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
-import java.util.HashMap;
-import java.util.List;
-import java.util.Map;
import static org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManagerConfig.LOG_DIR;
-@SuppressWarnings("deprecation") // Added for Scala 2.12 compatibility for usages of JavaConverters
+@ExtendWith(value = ClusterTestExtensions.class)
+@Tag("integration")
public class TopicBasedRemoteLogMetadataManagerRestartTest {
private static final int SEG_SIZE = 1024 * 1024;
private final Time time = new MockTime(1);
private final String logDir = TestUtils.tempDirectory("_rlmm_segs_").getAbsolutePath();
+ private final ClusterInstance clusterInstance;
- private TopicBasedRemoteLogMetadataManagerHarness remoteLogMetadataManagerHarness;
-
- @BeforeEach
- public void setup() {
- // Start the cluster and initialize TopicBasedRemoteLogMetadataManager.
- remoteLogMetadataManagerHarness = new TopicBasedRemoteLogMetadataManagerHarness() {
- protected Map<String, Object> overrideRemoteLogMetadataManagerProps() {
- Map<String, Object> props = new HashMap<>();
- props.put(LOG_DIR, logDir);
- return props;
- }
- };
- remoteLogMetadataManagerHarness.initialize(Collections.emptySet(), true);
+ TopicBasedRemoteLogMetadataManagerRestartTest(ClusterInstance clusterInstance) { // Constructor injections
+ this.clusterInstance = clusterInstance;
}
- private void startTopicBasedRemoteLogMetadataManagerHarness(boolean startConsumerThread) {
- remoteLogMetadataManagerHarness.initializeRemoteLogMetadataManager(Collections.emptySet(), startConsumerThread, RemoteLogMetadataTopicPartitioner::new);
+ private TopicBasedRemoteLogMetadataManager createTopicBasedRemoteLogMetadataManager() {
+ return RemoteLogMetadataManagerTestUtils.builder()
+ .topicIdPartitions(Collections.emptySet())
+ .bootstrapServers(clusterInstance.bootstrapServers())
+ .startConsumerThread(true)
+ .remoteLogMetadataTopicPartitioner(RemoteLogMetadataTopicPartitioner::new)
+ .overrideRemoteLogMetadataManagerProps(Collections.singletonMap(LOG_DIR, logDir))
+ .build();
}
- @AfterEach
- public void teardown() throws IOException {
- if (remoteLogMetadataManagerHarness != null) {
- remoteLogMetadataManagerHarness.close();
- }
- }
-
- private void stopTopicBasedRemoteLogMetadataManagerHarness() {
- remoteLogMetadataManagerHarness.closeRemoteLogMetadataManager();
- }
-
- private TopicBasedRemoteLogMetadataManager topicBasedRlmm() {
- return remoteLogMetadataManagerHarness.remoteLogMetadataManager();
- }
-
- @Test
+ @ClusterTest(brokers = 3)
public void testRLMMAPIsAfterRestart() throws Exception {
// Create topics.
String leaderTopic = "new-leader";
- HashMap<Object, Seq<Object>> assignedLeaderTopicReplicas = new HashMap<>();
- List<Object> leaderTopicReplicas = new ArrayList<>();
- // Set broker id 0 as the first entry which is taken as the leader.
- leaderTopicReplicas.add(0);
- leaderTopicReplicas.add(1);
- leaderTopicReplicas.add(2);
- assignedLeaderTopicReplicas.put(0, JavaConverters.asScalaBuffer(leaderTopicReplicas));
- remoteLogMetadataManagerHarness.createTopicWithAssignment(
- leaderTopic, JavaConverters.mapAsScalaMap(assignedLeaderTopicReplicas),
- remoteLogMetadataManagerHarness.listenerName());
-
String followerTopic = "new-follower";
- HashMap<Object, Seq<Object>> assignedFollowerTopicReplicas = new HashMap<>();
- List<Object> followerTopicReplicas = new ArrayList<>();
- // Set broker id 1 as the first entry which is taken as the leader.
- followerTopicReplicas.add(1);
- followerTopicReplicas.add(2);
- followerTopicReplicas.add(0);
- assignedFollowerTopicReplicas.put(0, JavaConverters.asScalaBuffer(followerTopicReplicas));
- remoteLogMetadataManagerHarness.createTopicWithAssignment(followerTopic,
- JavaConverters.mapAsScalaMap(assignedFollowerTopicReplicas),
- remoteLogMetadataManagerHarness.listenerName());
+ try (Admin admin = clusterInstance.createAdminClient()) {
+ // Set broker id 0 as the first entry which is taken as the leader.
+ NewTopic newLeaderTopic = new NewTopic(leaderTopic, Collections.singletonMap(0, Arrays.asList(0, 1, 2)));
+ // Set broker id 1 as the first entry which is taken as the leader.
+ NewTopic newFollowerTopic = new NewTopic(followerTopic, Collections.singletonMap(0, Arrays.asList(1, 2, 0)));
+ admin.createTopics(Arrays.asList(newLeaderTopic, newFollowerTopic));
+ }
final TopicIdPartition leaderTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(leaderTopic, 0));
final TopicIdPartition followerTopicIdPartition = new TopicIdPartition(Uuid.randomUuid(), new TopicPartition(followerTopic, 0));
-
- // Register these partitions to RLMM.
- topicBasedRlmm().onPartitionLeadershipChanges(Collections.singleton(leaderTopicIdPartition), Collections.singleton(followerTopicIdPartition));
-
- // Add segments for these partitions, but they are not available as they have not yet been subscribed.
RemoteLogSegmentMetadata leaderSegmentMetadata = new RemoteLogSegmentMetadata(
new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- topicBasedRlmm().addRemoteLogSegmentMetadata(leaderSegmentMetadata).get();
-
RemoteLogSegmentMetadata followerSegmentMetadata = new RemoteLogSegmentMetadata(
new RemoteLogSegmentId(followerTopicIdPartition, Uuid.randomUuid()),
0, 100, -1L, 0,
time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 0L));
- topicBasedRlmm().addRemoteLogSegmentMetadata(followerSegmentMetadata).get();
-
- // Stop TopicBasedRemoteLogMetadataManager only.
- stopTopicBasedRemoteLogMetadataManagerHarness();
- // Start TopicBasedRemoteLogMetadataManager
- startTopicBasedRemoteLogMetadataManagerHarness(true);
+ try (TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ // Register these partitions to RemoteLogMetadataManager.
+ topicBasedRemoteLogMetadataManager.onPartitionLeadershipChanges(
+ Collections.singleton(leaderTopicIdPartition), Collections.singleton(followerTopicIdPartition));
- // Register these partitions to RLMM, which loads the respective metadata snapshots.
- topicBasedRlmm().onPartitionLeadershipChanges(
- Collections.singleton(leaderTopicIdPartition), Collections.singleton(followerTopicIdPartition));
-
- // Check for the stored entries from the earlier run.
- TestUtils.waitForCondition(() ->
- TestUtils.sameElementsWithoutOrder(Collections.singleton(leaderSegmentMetadata).iterator(),
- topicBasedRlmm().listRemoteLogSegments(leaderTopicIdPartition)),
- "Remote log segment metadata not available");
- TestUtils.waitForCondition(() ->
- TestUtils.sameElementsWithoutOrder(Collections.singleton(followerSegmentMetadata).iterator(),
- topicBasedRlmm().listRemoteLogSegments(followerTopicIdPartition)),
- "Remote log segment metadata not available");
- // Add one more segment
- RemoteLogSegmentMetadata leaderSegmentMetadata2 = new RemoteLogSegmentMetadata(
- new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()),
- 101, 200, -1L, 0,
- time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 101L));
- topicBasedRlmm().addRemoteLogSegmentMetadata(leaderSegmentMetadata2).get();
+ // Add segments for these partitions, but they are not available as they have not yet been subscribed.
+ topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(leaderSegmentMetadata).get();
+ topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(followerSegmentMetadata).get();
+ }
- // Check that both the stored segment and recently added segment are available.
- Assertions.assertTrue(TestUtils.sameElementsWithoutOrder(Arrays.asList(leaderSegmentMetadata, leaderSegmentMetadata2).iterator(),
- topicBasedRlmm().listRemoteLogSegments(leaderTopicIdPartition)));
+ try (TopicBasedRemoteLogMetadataManager topicBasedRemoteLogMetadataManager = createTopicBasedRemoteLogMetadataManager()) {
+ // Register these partitions to RemoteLogMetadataManager, which loads the respective metadata snapshots.
+ topicBasedRemoteLogMetadataManager.onPartitionLeadershipChanges(
+ Collections.singleton(leaderTopicIdPartition), Collections.singleton(followerTopicIdPartition));
+
+ // Check for the stored entries from the earlier run.
+ TestUtils.waitForCondition(() ->
+ TestUtils.sameElementsWithoutOrder(Collections.singleton(leaderSegmentMetadata).iterator(),
+ topicBasedRemoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition)),
+ "Remote log segment metadata not available");
+ TestUtils.waitForCondition(() ->
+ TestUtils.sameElementsWithoutOrder(Collections.singleton(followerSegmentMetadata).iterator(),
+ topicBasedRemoteLogMetadataManager.listRemoteLogSegments(followerTopicIdPartition)),
+ "Remote log segment metadata not available");
+ // Add one more segment
+ RemoteLogSegmentMetadata leaderSegmentMetadata2 = new RemoteLogSegmentMetadata(
+ new RemoteLogSegmentId(leaderTopicIdPartition, Uuid.randomUuid()),
+ 101, 200, -1L, 0,
+ time.milliseconds(), SEG_SIZE, Collections.singletonMap(0, 101L));
+ topicBasedRemoteLogMetadataManager.addRemoteLogSegmentMetadata(leaderSegmentMetadata2).get();
+
+ // Check that both the stored segment and recently added segment are available.
+ Assertions.assertTrue(TestUtils.sameElementsWithoutOrder(Arrays.asList(leaderSegmentMetadata, leaderSegmentMetadata2).iterator(),
+ topicBasedRemoteLogMetadataManager.listRemoteLogSegments(leaderTopicIdPartition)));
+ }
}
}
\ No newline at end of file
(kafka) 12/12: KAFKA-15776: Support added to update remote.fetch.max.wait.ms dynamically (#16203)
Posted by sa...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
satishd pushed a commit to branch 3.8
in repository https://gitbox.apache.org/repos/asf/kafka.git
commit d94a28b4a4f94143f51d71e03f63907cccc8a2e1
Author: Kamal Chandraprakash <kc...@uber.com>
AuthorDate: Mon Jun 10 20:42:12 2024 +0530
KAFKA-15776: Support added to update remote.fetch.max.wait.ms dynamically (#16203)
Reviewers: Satish Duggana <sa...@apache.org>, Luke Chen <sh...@gmail.com>
---
.../scala/kafka/server/DynamicBrokerConfig.scala | 3 +-
core/src/main/scala/kafka/server/KafkaConfig.scala | 2 ++
.../main/scala/kafka/server/ReplicaManager.scala | 3 +-
.../kafka/server/DynamicBrokerConfigTest.scala | 33 ++++++++++++++++++++++
.../log/remote/storage/RemoteLogManagerConfig.java | 4 ---
5 files changed, 38 insertions(+), 7 deletions(-)
diff --git a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
index 6fa43b560dd..9b75425666f 100755
--- a/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
+++ b/core/src/main/scala/kafka/server/DynamicBrokerConfig.scala
@@ -1202,6 +1202,7 @@ class DynamicRemoteLogConfig(server: KafkaBroker) extends BrokerReconfigurable w
object DynamicRemoteLogConfig {
val ReconfigurableConfigs = Set(
- RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP
+ RemoteLogManagerConfig.REMOTE_LOG_INDEX_FILE_CACHE_TOTAL_SIZE_BYTES_PROP,
+ RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP
)
}
diff --git a/core/src/main/scala/kafka/server/KafkaConfig.scala b/core/src/main/scala/kafka/server/KafkaConfig.scala
index 06bce64f623..745c8648e38 100755
--- a/core/src/main/scala/kafka/server/KafkaConfig.scala
+++ b/core/src/main/scala/kafka/server/KafkaConfig.scala
@@ -1211,6 +1211,8 @@ class KafkaConfig private(doLog: Boolean, val props: java.util.Map[_, _], dynami
def logLocalRetentionMs: java.lang.Long = getLong(RemoteLogManagerConfig.LOG_LOCAL_RETENTION_MS_PROP)
+ def remoteFetchMaxWaitMs = getInt(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP)
+
validateValues()
@nowarn("cat=deprecation")
diff --git a/core/src/main/scala/kafka/server/ReplicaManager.scala b/core/src/main/scala/kafka/server/ReplicaManager.scala
index 17e9a5c1b9e..6ab67a84ec4 100644
--- a/core/src/main/scala/kafka/server/ReplicaManager.scala
+++ b/core/src/main/scala/kafka/server/ReplicaManager.scala
@@ -1479,10 +1479,9 @@ class ReplicaManager(val config: KafkaConfig,
return Some(createLogReadResult(e))
}
- val remoteFetchMaxWaitMs = config.remoteLogManagerConfig.remoteFetchMaxWaitMs()
+ val remoteFetchMaxWaitMs = config.remoteFetchMaxWaitMs.toLong
val remoteFetch = new DelayedRemoteFetch(remoteFetchTask, remoteFetchResult, remoteFetchInfo, remoteFetchMaxWaitMs,
fetchPartitionStatus, params, logReadResults, this, responseCallback)
-
delayedRemoteFetchPurgatory.tryCompleteElseWatch(remoteFetch, Seq(key))
None
}
diff --git a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
index 328d6e41b57..2e5d77cdc6f 100755
--- a/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
+++ b/core/src/test/scala/unit/kafka/server/DynamicBrokerConfigTest.scala
@@ -792,6 +792,39 @@ class DynamicBrokerConfigTest {
verifyIncorrectLogLocalRetentionProps(2000L, 1000L, -1, 100)
}
+ @Test
+ def testDynamicRemoteFetchMaxWaitMsConfig(): Unit = {
+ val props = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
+ val config = KafkaConfig(props)
+ val kafkaBroker = mock(classOf[KafkaBroker])
+ when(kafkaBroker.config).thenReturn(config)
+ assertEquals(500, config.remoteFetchMaxWaitMs)
+
+ val dynamicRemoteLogConfig = new DynamicRemoteLogConfig(kafkaBroker)
+ config.dynamicConfig.initialize(None, None)
+ config.dynamicConfig.addBrokerReconfigurable(dynamicRemoteLogConfig)
+
+ val newProps = new Properties()
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "30000")
+ // update default config
+ config.dynamicConfig.validate(newProps, perBrokerConfig = false)
+ config.dynamicConfig.updateDefaultConfig(newProps)
+ assertEquals(30000, config.remoteFetchMaxWaitMs)
+
+ // update per broker config
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, "10000")
+ config.dynamicConfig.validate(newProps, perBrokerConfig = true)
+ config.dynamicConfig.updateBrokerConfig(0, newProps)
+ assertEquals(10000, config.remoteFetchMaxWaitMs)
+
+ // invalid values
+ for (maxWaitMs <- Seq(-1, 0)) {
+ newProps.put(RemoteLogManagerConfig.REMOTE_FETCH_MAX_WAIT_MS_PROP, maxWaitMs.toString)
+ assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(newProps, perBrokerConfig = true))
+ assertThrows(classOf[ConfigException], () => config.dynamicConfig.validate(newProps, perBrokerConfig = false))
+ }
+ }
+
@Test
def testUpdateDynamicRemoteLogManagerConfig(): Unit = {
val origProps = TestUtils.createBrokerConfig(0, TestUtils.MockZkConnect, port = 8181)
diff --git a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
index 6323ba65bfe..25f1334226a 100644
--- a/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
+++ b/storage/src/main/java/org/apache/kafka/server/log/remote/storage/RemoteLogManagerConfig.java
@@ -471,10 +471,6 @@ public final class RemoteLogManagerConfig extends AbstractConfig {
return getInt(REMOTE_LOG_MANAGER_FETCH_QUOTA_WINDOW_SIZE_SECONDS_PROP);
}
- public int remoteFetchMaxWaitMs() {
- return getInt(REMOTE_FETCH_MAX_WAIT_MS_PROP);
- }
-
public static void main(String[] args) {
System.out.println(configDef().toHtml(4, config -> "remote_log_manager_" + config));
}