Skip to content

Commit 8b749cc

Browse files
TaiJuWuTaiJuWu
committed
remove common test dependency
1 parent 6a8b0bb commit 8b749cc

File tree

3 files changed

+88
-104
lines changed

3 files changed

+88
-104
lines changed

‎clients/src/test/java/org/apache/kafka/test/TestUtils.java‎

Lines changed: 87 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.apache.kafka.common.Node;
2424
import org.apache.kafka.common.PartitionInfo;
2525
import org.apache.kafka.common.TopicPartition;
26+
import org.apache.kafka.common.compress.Compression;
2627
import org.apache.kafka.common.feature.Features;
2728
import org.apache.kafka.common.feature.SupportedVersionRange;
2829
import org.apache.kafka.common.message.ApiMessageType;
@@ -31,6 +32,12 @@
3132
import org.apache.kafka.common.network.Send;
3233
import org.apache.kafka.common.protocol.ApiKeys;
3334
import org.apache.kafka.common.protocol.Errors;
35+
import org.apache.kafka.common.record.DefaultRecordBatch;
36+
import org.apache.kafka.common.record.MemoryRecords;
37+
import org.apache.kafka.common.record.MemoryRecordsBuilder;
38+
import org.apache.kafka.common.record.RecordBatch;
39+
import org.apache.kafka.common.record.SimpleRecord;
40+
import org.apache.kafka.common.record.TimestampType;
3441
import org.apache.kafka.common.record.UnalignedRecords;
3542
import org.apache.kafka.common.requests.ApiVersionsResponse;
3643
import org.apache.kafka.common.requests.ByteBufferChannel;
@@ -742,4 +749,84 @@ public static ApiVersionsResponse createApiVersionsResponse(
742749
setZkMigrationEnabled(zkMigrationEnabled).
743750
build();
744751
}
752+
753+
public static MemoryRecords singletonRecords(byte[] value, byte[] key) {
754+
return singletonRecords(value, key, Compression.NONE, RecordBatch.NO_TIMESTAMP, RecordBatch.CURRENT_MAGIC_VALUE);
755+
}
756+
757+
public static MemoryRecords singletonRecords(byte[] value, long timestamp) {
758+
return singletonRecords(value, null, Compression.NONE, timestamp, RecordBatch.CURRENT_MAGIC_VALUE);
759+
}
760+
761+
public static MemoryRecords singletonRecords(
762+
byte[] value
763+
) {
764+
return records(List.of(new SimpleRecord(RecordBatch.NO_TIMESTAMP, null, value)),
765+
RecordBatch.CURRENT_MAGIC_VALUE,
766+
Compression.NONE,
767+
RecordBatch.NO_PRODUCER_ID,
768+
RecordBatch.NO_PRODUCER_EPOCH,
769+
RecordBatch.NO_SEQUENCE,
770+
0,
771+
RecordBatch.NO_PARTITION_LEADER_EPOCH
772+
);
773+
}
774+
775+
public static MemoryRecords singletonRecords(
776+
byte[] value,
777+
byte[] key,
778+
Compression codec,
779+
long timestamp,
780+
byte magicValue
781+
) {
782+
return records(List.of(new SimpleRecord(timestamp, key, value)),
783+
magicValue, codec,
784+
RecordBatch.NO_PRODUCER_ID,
785+
RecordBatch.NO_PRODUCER_EPOCH,
786+
RecordBatch.NO_SEQUENCE,
787+
0,
788+
RecordBatch.NO_PARTITION_LEADER_EPOCH
789+
);
790+
}
791+
792+
public static MemoryRecords singletonRecords(byte[] value, byte[] key, long timestamp) {
793+
return singletonRecords(value, key, Compression.NONE, timestamp, RecordBatch.CURRENT_MAGIC_VALUE);
794+
}
795+
796+
public static MemoryRecords records(List<SimpleRecord> records) {
797+
return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
798+
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, 0L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
799+
}
800+
801+
public static MemoryRecords records(List<SimpleRecord> records, long baseOffset) {
802+
return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
803+
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, baseOffset, RecordBatch.NO_PARTITION_LEADER_EPOCH);
804+
}
805+
806+
public static MemoryRecords records(List<SimpleRecord> records, long baseOffset, int partitionLeaderEpoch) {
807+
return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
808+
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, baseOffset, partitionLeaderEpoch);
809+
}
810+
811+
public static MemoryRecords records(List<SimpleRecord> records, byte magicValue, Compression compression) {
812+
return records(records, magicValue, compression, RecordBatch.NO_PRODUCER_ID,
813+
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, 0L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
814+
}
815+
816+
public static MemoryRecords records(List<SimpleRecord> records,
817+
byte magicValue,
818+
Compression compression,
819+
long producerId,
820+
short producerEpoch,
821+
int sequence,
822+
long baseOffset,
823+
int partitionLeaderEpoch) {
824+
ByteBuffer buf = ByteBuffer.allocate(DefaultRecordBatch.sizeInBytes(records));
825+
MemoryRecordsBuilder builder = MemoryRecords.builder(buf, magicValue, compression, TimestampType.CREATE_TIME, baseOffset,
826+
System.currentTimeMillis(), producerId, producerEpoch, sequence, false, partitionLeaderEpoch);
827+
for (SimpleRecord record : records) {
828+
builder.append(record);
829+
}
830+
return builder.build();
831+
}
745832
}

‎storage/src/test/java/org/apache/kafka/storage/internals/log/UnifiedLogTest.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -24,14 +24,14 @@
2424
import org.apache.kafka.common.record.MemoryRecords;
2525
import org.apache.kafka.common.record.RecordBatch;
2626
import org.apache.kafka.common.record.SimpleRecord;
27-
import org.apache.kafka.common.test.TestUtils;
2827
import org.apache.kafka.common.utils.Utils;
2928
import org.apache.kafka.coordinator.transaction.TransactionLogConfig;
3029
import org.apache.kafka.server.storage.log.FetchIsolation;
3130
import org.apache.kafka.server.util.MockTime;
3231
import org.apache.kafka.server.util.Scheduler;
3332
import org.apache.kafka.storage.internals.epoch.LeaderEpochFileCache;
3433
import org.apache.kafka.storage.log.metrics.BrokerTopicStats;
34+
import org.apache.kafka.test.TestUtils;
3535

3636
import org.junit.jupiter.api.AfterEach;
3737
import org.junit.jupiter.api.Test;

‎test-common/test-common-runtime/src/main/java/org/apache/kafka/common/test/TestUtils.java‎

Lines changed: 0 additions & 103 deletions
Original file line numberDiff line numberDiff line change
@@ -16,14 +16,7 @@
1616
*/
1717
package org.apache.kafka.common.test;
1818

19-
import org.apache.kafka.common.compress.Compression;
2019
import org.apache.kafka.common.errors.TimeoutException;
21-
import org.apache.kafka.common.record.DefaultRecordBatch;
22-
import org.apache.kafka.common.record.MemoryRecords;
23-
import org.apache.kafka.common.record.MemoryRecordsBuilder;
24-
import org.apache.kafka.common.record.RecordBatch;
25-
import org.apache.kafka.common.record.SimpleRecord;
26-
import org.apache.kafka.common.record.TimestampType;
2720
import org.apache.kafka.common.utils.Exit;
2821
import org.apache.kafka.common.utils.Utils;
2922

@@ -32,9 +25,7 @@
3225

3326
import java.io.File;
3427
import java.io.IOException;
35-
import java.nio.ByteBuffer;
3628
import java.nio.file.Files;
37-
import java.util.List;
3829
import java.util.Random;
3930
import java.util.function.Supplier;
4031

@@ -57,7 +48,6 @@ class TestUtils {
5748

5849
private static final long DEFAULT_POLL_INTERVAL_MS = 100;
5950
private static final long DEFAULT_MAX_WAIT_MS = 15_000;
60-
private static final Random RANDOM = new Random();
6151

6252
/**
6353
* Create an empty file in the default temporary-file directory, using `kafka` as the prefix and `tmp` as the
@@ -155,97 +145,4 @@ public static void waitForCondition(final Supplier<Boolean> testCondition,
155145
String conditionDetails) throws InterruptedException {
156146
waitForCondition(testCondition, maxWaitMs, () -> conditionDetails);
157147
}
158-
159-
public static File randomPartitionLogDir(File parentDir) {
160-
int attempts = 1000;
161-
while (attempts > 0) {
162-
File f = new File(parentDir, "kafka-" + RANDOM.nextInt(1000000));
163-
if (f.mkdir()) {
164-
f.deleteOnExit();
165-
return f;
166-
}
167-
attempts--;
168-
}
169-
throw new RuntimeException("Failed to create directory after 1000 attempts");
170-
}
171-
172-
public static MemoryRecords singletonRecords(byte[] value, byte[] key) {
173-
return singletonRecords(value, key, Compression.NONE, RecordBatch.NO_TIMESTAMP, RecordBatch.CURRENT_MAGIC_VALUE);
174-
}
175-
176-
public static MemoryRecords singletonRecords(byte[] value, long timestamp) {
177-
return singletonRecords(value, null, Compression.NONE, timestamp, RecordBatch.CURRENT_MAGIC_VALUE);
178-
}
179-
180-
public static MemoryRecords singletonRecords(
181-
byte[] value
182-
) {
183-
return records(List.of(new SimpleRecord(RecordBatch.NO_TIMESTAMP, null, value)),
184-
RecordBatch.CURRENT_MAGIC_VALUE,
185-
Compression.NONE,
186-
RecordBatch.NO_PRODUCER_ID,
187-
RecordBatch.NO_PRODUCER_EPOCH,
188-
RecordBatch.NO_SEQUENCE,
189-
0,
190-
RecordBatch.NO_PARTITION_LEADER_EPOCH
191-
);
192-
}
193-
194-
public static MemoryRecords singletonRecords(
195-
byte[] value,
196-
byte[] key,
197-
Compression codec,
198-
long timestamp,
199-
byte magicValue
200-
) {
201-
return records(List.of(new SimpleRecord(timestamp, key, value)),
202-
magicValue, codec,
203-
RecordBatch.NO_PRODUCER_ID,
204-
RecordBatch.NO_PRODUCER_EPOCH,
205-
RecordBatch.NO_SEQUENCE,
206-
0,
207-
RecordBatch.NO_PARTITION_LEADER_EPOCH
208-
);
209-
}
210-
211-
public static MemoryRecords singletonRecords(byte[] value, byte[] key, long timestamp) {
212-
return singletonRecords(value, key, Compression.NONE, timestamp, RecordBatch.CURRENT_MAGIC_VALUE);
213-
}
214-
215-
public static MemoryRecords records(List<SimpleRecord> records) {
216-
return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
217-
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, 0L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
218-
}
219-
220-
public static MemoryRecords records(List<SimpleRecord> records, long baseOffset) {
221-
return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
222-
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, baseOffset, RecordBatch.NO_PARTITION_LEADER_EPOCH);
223-
}
224-
225-
public static MemoryRecords records(List<SimpleRecord> records, long baseOffset, int partitionLeaderEpoch) {
226-
return records(records, RecordBatch.CURRENT_MAGIC_VALUE, Compression.NONE, RecordBatch.NO_PRODUCER_ID,
227-
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, baseOffset, partitionLeaderEpoch);
228-
}
229-
230-
public static MemoryRecords records(List<SimpleRecord> records, byte magicValue, Compression compression) {
231-
return records(records, magicValue, compression, RecordBatch.NO_PRODUCER_ID,
232-
RecordBatch.NO_PRODUCER_EPOCH, RecordBatch.NO_SEQUENCE, 0L, RecordBatch.NO_PARTITION_LEADER_EPOCH);
233-
}
234-
235-
public static MemoryRecords records(List<SimpleRecord> records,
236-
byte magicValue,
237-
Compression compression,
238-
long producerId,
239-
short producerEpoch,
240-
int sequence,
241-
long baseOffset,
242-
int partitionLeaderEpoch) {
243-
ByteBuffer buf = ByteBuffer.allocate(DefaultRecordBatch.sizeInBytes(records));
244-
MemoryRecordsBuilder builder = MemoryRecords.builder(buf, magicValue, compression, TimestampType.CREATE_TIME, baseOffset,
245-
System.currentTimeMillis(), producerId, producerEpoch, sequence, false, partitionLeaderEpoch);
246-
for (SimpleRecord record : records) {
247-
builder.append(record);
248-
}
249-
return builder.build();
250-
}
251148
}

0 commit comments

Comments
 (0)