Skip to content

Commit 8c079f4

Browse files
m1a2stm1a2st
authored
KAFKA-17689 Migrate TieredStorageTestHarness to new test infra (AlterLogDirTest) (apache#22306)
Migrate `AlterLogDirTest` to the new test infrastructure Reviewers: Chia-Ping Tsai <chia7712@gmail.com>, PoAn Yang <payang@apache.org>, TaiJuWu <tjwu1217@gmail.com>
1 parent a7dfb89 commit 8c079f4

3 files changed

Lines changed: 72 additions & 24 deletions

File tree

‎storage/src/test/java/org/apache/kafka/tiered/storage/integration/AlterLogDirTest.java‎

Lines changed: 51 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -16,24 +16,53 @@
1616
*/
1717
package org.apache.kafka.tiered.storage.integration;
1818

19+
import org.apache.kafka.clients.consumer.ConsumerConfig;
20+
import org.apache.kafka.clients.consumer.GroupProtocol;
21+
import org.apache.kafka.common.test.ClusterInstance;
22+
import org.apache.kafka.common.test.api.ClusterConfig;
23+
import org.apache.kafka.common.test.api.ClusterTemplate;
24+
import org.apache.kafka.common.test.api.Type;
25+
import org.apache.kafka.tiered.storage.TieredStorageTestAction;
1926
import org.apache.kafka.tiered.storage.TieredStorageTestBuilder;
20-
import org.apache.kafka.tiered.storage.TieredStorageTestHarness;
27+
import org.apache.kafka.tiered.storage.TieredStorageTestContext;
2128
import org.apache.kafka.tiered.storage.specs.KeyValueSpec;
2229

2330
import java.util.List;
31+
import java.util.Locale;
32+
import java.util.Map;
33+
import java.util.Set;
2434

2535
import static org.apache.kafka.common.utils.Utils.mkEntry;
2636
import static org.apache.kafka.common.utils.Utils.mkMap;
37+
import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.createServerPropsForRemoteStorage;
2738

28-
public final class AlterLogDirTest extends TieredStorageTestHarness {
39+
public final class AlterLogDirTest {
2940

30-
@Override
31-
public int brokerCount() {
32-
return 2;
41+
private static final int BROKER_COUNT = 3;
42+
43+
private static List<ClusterConfig> clusterConfig() {
44+
return List.of(ClusterConfig.defaultBuilder()
45+
.setTypes(Set.of(Type.KRAFT))
46+
.setBrokers(BROKER_COUNT)
47+
.setDisksPerBroker(2)
48+
.setServerProperties(createServerPropsForRemoteStorage(
49+
AlterLogDirTest.class.getSimpleName().toLowerCase(Locale.ROOT),
50+
BROKER_COUNT,
51+
5))
52+
.build());
53+
}
54+
55+
@ClusterTemplate("clusterConfig")
56+
public void testAlterLogDirWithClassicGroupProtocol(ClusterInstance clusterInstance) throws Exception {
57+
executeAlterLogDirTest(clusterInstance, GroupProtocol.CLASSIC);
3358
}
3459

35-
@Override
36-
protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
60+
@ClusterTemplate("clusterConfig")
61+
public void testAlterLogDirWithConsumerGroupProtocol(ClusterInstance clusterInstance) throws Exception {
62+
executeAlterLogDirTest(clusterInstance, GroupProtocol.CONSUMER);
63+
}
64+
65+
private void executeAlterLogDirTest(ClusterInstance clusterInstance, GroupProtocol groupProtocol) throws Exception {
3766
final String topicB = "topicB";
3867
final int p0 = 0;
3968
final int partitionCount = 1;
@@ -43,8 +72,9 @@ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
4372
final int broker0 = 0;
4473
final int broker1 = 1;
4574

75+
TieredStorageTestBuilder builder = new TieredStorageTestBuilder();
4676
builder
47-
// create topicB with 1 partition and 1 RF
77+
// create topicB with 1 partition and 2 RF
4878
.createTopic(topicB, partitionCount, replicationFactor, maxBatchCountPerSegment,
4979
mkMap(mkEntry(p0, List.of(broker1, broker0))), enableRemoteLogStorage)
5080
// send records to partition 0
@@ -63,5 +93,18 @@ protected void writeTestSpecifications(TieredStorageTestBuilder builder) {
6393
// consume from the beginning of the topic to read data from local and remote storage
6494
.expectFetchFromTieredStorage(broker0, topicB, p0, 3)
6595
.consume(topicB, p0, 0L, 4, 3);
96+
97+
Map<String, Object> extraConsumerProps = Map.of(
98+
ConsumerConfig.GROUP_PROTOCOL_CONFIG, groupProtocol.name().toLowerCase(Locale.ROOT)
99+
);
100+
try (TieredStorageTestContext context = new TieredStorageTestContext(clusterInstance, extraConsumerProps)) {
101+
try {
102+
for (TieredStorageTestAction action : builder.complete()) {
103+
action.execute(context);
104+
}
105+
} finally {
106+
context.printReport(System.out);
107+
}
108+
}
66109
}
67110
}

‎storage/src/test/java/org/apache/kafka/tiered/storage/integration/TransactionsWithTieredStoreTest.java‎

Lines changed: 2 additions & 16 deletions
Original file line numberDiff line numberDiff line change
@@ -23,24 +23,21 @@
2323
import org.apache.kafka.common.test.ClusterInstance;
2424
import org.apache.kafka.common.test.api.ClusterConfig;
2525
import org.apache.kafka.common.test.api.ClusterTemplate;
26-
import org.apache.kafka.common.test.api.TestKitDefaults;
2726
import org.apache.kafka.common.test.api.Type;
2827
import org.apache.kafka.server.common.Feature;
2928
import org.apache.kafka.server.config.ServerLogConfigs;
30-
import org.apache.kafka.server.log.remote.storage.RemoteLogManagerConfig;
3129
import org.apache.kafka.test.TestUtils;
3230
import org.apache.kafka.tiered.storage.integration.TransactionsTestHelper.TransactionHooks;
3331
import org.apache.kafka.tiered.storage.utils.BrokerLocalStorage;
3432

3533
import java.util.HashMap;
3634
import java.util.List;
3735
import java.util.Map;
38-
import java.util.Properties;
3936
import java.util.Set;
4037
import java.util.stream.Collectors;
4138

4239
import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.STORAGE_WAIT_TIMEOUT_SEC;
43-
import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.createPropsForRemoteStorage;
40+
import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.createServerPropsForRemoteStorage;
4441
import static org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.createTopicConfigForRemoteStorage;
4542

4643
/**
@@ -56,10 +53,7 @@ public class TransactionsWithTieredStoreTest {
5653
private static final int BROKER_COUNT = 3;
5754

5855
private static Map<String, String> baseServerProperties() {
59-
String storageDirPath = TestUtils.tempDirectory(
60-
"kafka-remote-tier-" + TEST_CLASS_NAME).getAbsolutePath();
61-
62-
Map<String, String> serverProps = new HashMap<>();
56+
Map<String, String> serverProps = createServerPropsForRemoteStorage(TEST_CLASS_NAME, BROKER_COUNT, 3);
6357
serverProps.put(ServerLogConfigs.AUTO_CREATE_TOPICS_ENABLE_CONFIG, "false");
6458
serverProps.put("offsets.topic.num.partitions", "1");
6559
serverProps.put("transaction.state.log.num.partitions", "3");
@@ -70,14 +64,6 @@ private static Map<String, String> baseServerProperties() {
7064
serverProps.put("auto.leader.rebalance.enable", "false");
7165
serverProps.put("group.initial.rebalance.delay.ms", "0");
7266
serverProps.put("transaction.abort.timed.out.transaction.cleanup.interval.ms", "200");
73-
74-
Properties tieredProps = createPropsForRemoteStorage(
75-
TEST_CLASS_NAME, storageDirPath, BROKER_COUNT, 3, new Properties());
76-
tieredProps.forEach((k, v) -> serverProps.put(k.toString(), v.toString()));
77-
78-
serverProps.put(RemoteLogManagerConfig.REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP,
79-
TestKitDefaults.DEFAULT_BROKER_LISTENER_NAME);
80-
8167
return serverProps;
8268
}
8369

‎storage/src/test/java/org/apache/kafka/tiered/storage/utils/TieredStorageTestUtils.java‎

Lines changed: 19 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -22,6 +22,7 @@
2222
import org.apache.kafka.common.TopicPartition;
2323
import org.apache.kafka.common.config.TopicConfig;
2424
import org.apache.kafka.common.record.internal.Record;
25+
import org.apache.kafka.common.test.api.TestKitDefaults;
2526
import org.apache.kafka.server.config.ServerConfigs;
2627
import org.apache.kafka.server.config.ServerLogConfigs;
2728
import org.apache.kafka.server.log.remote.metadata.storage.TopicBasedRemoteLogMetadataManager;
@@ -164,6 +165,24 @@ public static Properties createPropsForRemoteStorage(String testClassName,
164165
return overridingProps;
165166
}
166167

168+
public static Map<String, String> createServerPropsForRemoteStorage(
169+
String testClassName,
170+
int brokerCount,
171+
int numRemoteLogMetadataPartitions
172+
) {
173+
String storageDirPath = org.apache.kafka.test.TestUtils
174+
.tempDirectory("kafka-remote-tier-" + testClassName).getAbsolutePath();
175+
Properties tieredProps = createPropsForRemoteStorage(
176+
testClassName, storageDirPath, brokerCount, numRemoteLogMetadataPartitions, new Properties());
177+
Map<String, String> serverProps = new HashMap<>();
178+
tieredProps.forEach((k, v) -> serverProps.put(k.toString(), v.toString()));
179+
serverProps.put(
180+
REMOTE_LOG_METADATA_MANAGER_LISTENER_NAME_PROP,
181+
TestKitDefaults.DEFAULT_BROKER_LISTENER_NAME
182+
);
183+
return serverProps;
184+
}
185+
167186
public static Map<String, String> createTopicConfigForRemoteStorage(boolean enableRemoteStorage,
168187
int maxRecordBatchPerSegment) {
169188
Map<String, String> topicProps = new HashMap<>();

0 commit comments

Comments
 (0)