This is an automated email from the ASF dual-hosted git repository.
chia7712 pushed a commit to branch trunk
in repository https://gitbox.apache.org/repos/asf/kafka.git
The following commit(s) were added to refs/heads/trunk by this push:
new 094aa6ba949 MINOR: Remove hamcrest from storage (#22528)
094aa6ba949 is described below
commit 094aa6ba949686c857ffa28edbca1cb11da15b9f
Author: Mickael Maison <[email protected]>
AuthorDate: Sun Jun 14 09:07:34 2026 +0200
MINOR: Remove hamcrest from storage (#22528)
We just used hamcrest for a custom matcher, and it's easy to refactor it
to only use plain Java/JUnit.
Reviewers: Chia-Ping Tsai <[email protected]>
---
build.gradle | 1 -
checkstyle/import-control-storage.xml | 1 -
.../tiered/storage/actions/ConsumeAction.java | 5 +-
.../tiered/storage/actions/ProduceAction.java | 7 +-
.../storage/utils/RecordsKeyValueAssert.java | 135 ++++++++++++++++
.../storage/utils/RecordsKeyValueMatcher.java | 178 ---------------------
6 files changed, 140 insertions(+), 187 deletions(-)
diff --git a/build.gradle b/build.gradle
index d49828626ca..ad9c74bb1ba 100644
--- a/build.gradle
+++ b/build.gradle
@@ -2478,7 +2478,6 @@ project(':storage') {
testImplementation testFixtures(project(':server-common'))
testImplementation testFixtures(project(':metadata'))
testImplementation project(':transaction-coordinator')
- testImplementation libs.hamcrest
testImplementation libs.jacksonDataformatYaml
testImplementation libs.junitJupiter
testImplementation libs.mockitoCore
diff --git a/checkstyle/import-control-storage.xml
b/checkstyle/import-control-storage.xml
index 3599ca5fffe..7f6f56dd16a 100644
--- a/checkstyle/import-control-storage.xml
+++ b/checkstyle/import-control-storage.xml
@@ -27,7 +27,6 @@
<allow pkg="javax.management" />
<allow pkg="org.slf4j" />
<allow pkg="org.junit" />
- <allow pkg="org.hamcrest" />
<allow pkg="org.mockito" />
<allow pkg="org.opentest4j" />
<allow pkg="java.security" />
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ConsumeAction.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ConsumeAction.java
index 580b2324b21..5446fcac580 100644
---
a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ConsumeAction.java
+++
b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ConsumeAction.java
@@ -38,9 +38,8 @@ import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent
import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.FETCH_SEGMENT;
import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.FETCH_TIME_INDEX;
import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.FETCH_TRANSACTION_INDEX;
-import static
org.apache.kafka.tiered.storage.utils.RecordsKeyValueMatcher.correspondTo;
+import static
org.apache.kafka.tiered.storage.utils.RecordsKeyValueAssert.assertRecordsCorrespondTo;
import static
org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.tieredStorageRecords;
-import static org.hamcrest.MatcherAssert.assertThat;
import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertTrue;
@@ -124,7 +123,7 @@ public final class ConsumeAction implements
TieredStorageTestAction {
tieredStorageRecords.subList(indexOfFetchOffsetInTieredStorage,
tieredStorageRecords.size());
List<ConsumerRecord<String, String>> readRecords =
consumedRecords.subList(0, expectedFromSecondTierCount);
- assertThat(storedRecords, correspondTo(readRecords, topicPartition,
serde, serde));
+ assertRecordsCorrespondTo(storedRecords, readRecords, topicPartition,
serde, serde);
// (B) Assessment of the interactions between the source broker and
the second-tier storage.
for (LocalTieredStorageEvent.EventType eventType :
List.of(FETCH_SEGMENT, FETCH_OFFSET_INDEX, FETCH_TIME_INDEX,
FETCH_TRANSACTION_INDEX)) {
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ProduceAction.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ProduceAction.java
index b63ff1cd53a..3f435725247 100644
---
a/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ProduceAction.java
+++
b/storage/src/test/java/org/apache/kafka/tiered/storage/actions/ProduceAction.java
@@ -38,9 +38,8 @@ import java.util.concurrent.TimeoutException;
import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageCondition.expectEvent;
import static
org.apache.kafka.server.log.remote.storage.LocalTieredStorageEvent.EventType.COPY_SEGMENT;
-import static
org.apache.kafka.tiered.storage.utils.RecordsKeyValueMatcher.correspondTo;
+import static
org.apache.kafka.tiered.storage.utils.RecordsKeyValueAssert.assertRecordsCorrespondTo;
import static
org.apache.kafka.tiered.storage.utils.TieredStorageTestUtils.tieredStorageRecords;
-import static org.hamcrest.MatcherAssert.assertThat;
public final class ProduceAction implements TieredStorageTestAction {
@@ -128,7 +127,7 @@ public final class ProduceAction implements
TieredStorageTestAction {
// Verify that the produced records can be consumed from the
topic-partition.
List<ConsumerRecord<String, String>> consumedRecords =
context.consume(topicPartition, recordsToProduce.size(),
startOffset);
- assertThat(consumedRecords, correspondTo(recordsToProduce,
topicPartition, serde, serde));
+ assertRecordsCorrespondTo(recordsToProduce, consumedRecords,
topicPartition, serde, serde);
// Take a physical snapshot of the second-tier storage, and compare
the records found with
// those of the expected log segments.
@@ -160,6 +159,6 @@ public final class ProduceAction implements
TieredStorageTestAction {
private void compareRecords(List<Record> discoveredRecords,
List<ProducerRecord<String, String>>
producerRecords,
TopicPartition topicPartition) {
- assertThat(discoveredRecords, correspondTo(producerRecords,
topicPartition, serde, serde));
+ assertRecordsCorrespondTo(producerRecords, discoveredRecords,
topicPartition, serde, serde);
}
}
\ No newline at end of file
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/utils/RecordsKeyValueAssert.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/utils/RecordsKeyValueAssert.java
new file mode 100644
index 00000000000..651905b73d8
--- /dev/null
+++
b/storage/src/test/java/org/apache/kafka/tiered/storage/utils/RecordsKeyValueAssert.java
@@ -0,0 +1,135 @@
+/*
+ * 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.tiered.storage.utils;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.producer.ProducerRecord;
+import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.header.Header;
+import org.apache.kafka.common.record.internal.Record;
+import org.apache.kafka.common.record.internal.RecordBatch;
+import org.apache.kafka.common.record.internal.SimpleRecord;
+import org.apache.kafka.common.serialization.Deserializer;
+import org.apache.kafka.common.serialization.Serde;
+import org.apache.kafka.common.utils.Utils;
+
+import java.nio.ByteBuffer;
+import java.util.Collection;
+import java.util.Iterator;
+import java.util.Objects;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.fail;
+
+/**
+ * Assertion utility for comparing heterogeneous record types:
+ * {@link ProducerRecord}, {@link ConsumerRecord} or {@link Record}.
+ *
+ * <p>It is conceptually incorrect to try to match records of different
natures.
+ * Only a committed {@link Record} is univoque, whereas a {@link
ProducerRecord} or {@link ConsumerRecord} is
+ * a physical representation of a record-to-be or viewed record.
+ *
+ * <p>This utility breaches that semantic so that testers can avoid performing
manual comparisons on
+ * targeted internal fields of these type of records. This implementation only
compares key and
+ * value of the records.
+ */
+public final class RecordsKeyValueAssert {
+
+ private RecordsKeyValueAssert() {}
+
+ /**
+ * Asserts that the actual records correspond to the expected records by
comparing keys and values in order.
+ *
+ * @param <R1> The type of records used to formulate the expectations.
+ * @param <R2> The type of records compared against the expectations.
+ * @param <K> The type of the record keys.
+ * @param <V> The type of the record values.
+ * @param expectedRecords The records expected.
+ * @param actualRecords The records to verify.
+ * @param topicPartition The topic-partition which the records belong to.
+ * @param keySerde The {@link Serde} for the keys of the records.
+ * @param valueSerde The {@link Serde} for the values of the records.
+ */
+ public static <R1, R2, K, V> void assertRecordsCorrespondTo(Collection<R1>
expectedRecords,
+ Collection<R2>
actualRecords,
+ TopicPartition
topicPartition,
+ Serde<K>
keySerde,
+ Serde<V>
valueSerde) {
+ assertEquals(expectedRecords.size(), actualRecords.size());
+
+ Deserializer<K> keyDeserializer = keySerde.deserializer();
+ Deserializer<V> valueDeserializer = valueSerde.deserializer();
+ Iterator<R1> expectedIterator = expectedRecords.iterator();
+ Iterator<R2> actualIterator = actualRecords.iterator();
+ int index = 0;
+ while (expectedIterator.hasNext() && actualIterator.hasNext()) {
+ R1 expected = expectedIterator.next();
+ R2 actual = actualIterator.next();
+ SimpleRecord expectedRecord = convert(expected, topicPartition,
keySerde, valueSerde);
+ SimpleRecord actualRecord = convert(actual, topicPartition,
keySerde, valueSerde);
+ assertFieldEquals(expectedRecord.key(), actualRecord.key(),
+ keyDeserializer, "Record key", index, topicPartition);
+ assertFieldEquals(expectedRecord.value(), actualRecord.value(),
+ valueDeserializer, "Record value", index, topicPartition);
+ index++;
+ }
+ }
+
+ private static void assertFieldEquals(ByteBuffer expected,
+ ByteBuffer actual,
+ Deserializer<?> deserializer,
+ String fieldName,
+ int index,
+ TopicPartition topicPartition) {
+ if (!Objects.equals(expected, actual)) {
+ Object expectedDeserialized =
deserializer.deserialize(topicPartition.topic(),
+ Utils.toNullableArray(expected));
+ Object actualDeserialized =
deserializer.deserialize(topicPartition.topic(),
+ Utils.toNullableArray(actual));
+ fail(fieldName + " mismatch at index " + index + " for " +
topicPartition
+ + ". Expected: " + expectedDeserialized + ", Actual: " +
actualDeserialized);
+ }
+ }
+
+ @SuppressWarnings("unchecked")
+ private static <K, V> SimpleRecord convert(Object recordCandidate,
+ TopicPartition topicPartition,
+ Serde<K> keySerde,
+ Serde<V> valueSerde) {
+ if (recordCandidate instanceof ProducerRecord<?, ?> record) {
+ long timestamp = record.timestamp() != null ? record.timestamp() :
RecordBatch.NO_TIMESTAMP;
+ ByteBuffer keyBytes =
+
Utils.wrapNullable(keySerde.serializer().serialize(topicPartition.topic(), (K)
record.key()));
+ ByteBuffer valueBytes =
+
Utils.wrapNullable(valueSerde.serializer().serialize(topicPartition.topic(),
(V) record.value()));
+ Header[] headers = record.headers() != null ?
record.headers().toArray() : Record.EMPTY_HEADERS;
+ return new SimpleRecord(timestamp, keyBytes, valueBytes, headers);
+ } else if (recordCandidate instanceof ConsumerRecord<?, ?> record) {
+ ByteBuffer keyBytes =
+
Utils.wrapNullable(keySerde.serializer().serialize(topicPartition.topic(), (K)
record.key()));
+ ByteBuffer valueBytes =
+
Utils.wrapNullable(valueSerde.serializer().serialize(topicPartition.topic(),
(V) record.value()));
+ Header[] headers = record.headers() != null ?
record.headers().toArray() : Record.EMPTY_HEADERS;
+ return new SimpleRecord(record.timestamp(), keyBytes, valueBytes,
headers);
+ } else if (recordCandidate instanceof Record record) {
+ return new SimpleRecord(record.timestamp(), record.key(),
record.value(), record.headers());
+ } else {
+ throw new IllegalArgumentException("Unsupported record type: " +
recordCandidate.getClass().getSimpleName()
+ + ". Supported types: ProducerRecord, ConsumerRecord,
Record");
+ }
+ }
+}
diff --git
a/storage/src/test/java/org/apache/kafka/tiered/storage/utils/RecordsKeyValueMatcher.java
b/storage/src/test/java/org/apache/kafka/tiered/storage/utils/RecordsKeyValueMatcher.java
deleted file mode 100644
index 07e3b001f77..00000000000
---
a/storage/src/test/java/org/apache/kafka/tiered/storage/utils/RecordsKeyValueMatcher.java
+++ /dev/null
@@ -1,178 +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.tiered.storage.utils;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.header.Header;
-import org.apache.kafka.common.record.internal.Record;
-import org.apache.kafka.common.record.internal.RecordBatch;
-import org.apache.kafka.common.record.internal.SimpleRecord;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.apache.kafka.common.serialization.Serde;
-import org.apache.kafka.common.utils.Utils;
-
-import org.hamcrest.Description;
-import org.hamcrest.TypeSafeDiagnosingMatcher;
-
-import java.nio.ByteBuffer;
-import java.util.Collection;
-import java.util.Iterator;
-
-/**
- * @param <R1> The type of records used to formulate the expectations.
- * @param <R2> The type of records compared against the expectations.
- * @param <K> The type of the record keys.
- * @param <V> The type of the record values.
- */
-public final class RecordsKeyValueMatcher<R1, R2, K, V> extends
TypeSafeDiagnosingMatcher<Collection<R2>> {
-
- private final Collection<R1> expectedRecords;
- private final TopicPartition topicPartition;
- private final Serde<K> keySerde;
- private final Serde<V> valueSerde;
-
- /**
- * Heterogeneous matcher between alternative types of records:
- * {@link ProducerRecord}, {@link ConsumerRecord} or {@link Record}.
- *
- * It is conceptually incorrect to try to match records of different
natures.
- * Only a committed {@link Record} is univoque, whereas a {@link
ProducerRecord} or {@link ConsumerRecord} is
- * a physical representation of a record-to-be or viewed record.
- *
- * This matcher breaches that semantic so that testers can avoid
performing manual comparisons on
- * targeted internal fields of these type of records. This implementation
only compares key and
- * value of the records.
- *
- * @param expectedRecords The records expected.
- * @param topicPartition The topic-partition which the records belong to.
- * @param keySerde The {@link Serde} for the keys of the records.
- * @param valueSerde The {@link Serde} for the values of the records.
- */
- public RecordsKeyValueMatcher(Collection<R1> expectedRecords,
- TopicPartition topicPartition,
- Serde<K> keySerde,
- Serde<V> valueSerde) {
- this.expectedRecords = expectedRecords;
- this.topicPartition = topicPartition;
- this.keySerde = keySerde;
- this.valueSerde = valueSerde;
- }
-
- @Override
- public void describeTo(Description description) {
- description.appendText("Records of
").appendValue(topicPartition).appendText(": ").appendValue(expectedRecords);
- }
-
- @Override
- protected boolean matchesSafely(Collection<R2> actualRecords, Description
mismatchDescription) {
- if (expectedRecords.size() != actualRecords.size()) {
- mismatchDescription.appendText("Number of records differ.
Expected: ")
- .appendValue(expectedRecords.size())
- .appendText(", Actual: ")
- .appendValue(actualRecords.size())
- .appendText("; ");
- return false;
- }
- Iterator<R1> expectedIterator = expectedRecords.iterator();
- Iterator<R2> actualIterator = actualRecords.iterator();
- while (expectedIterator.hasNext() && actualIterator.hasNext()) {
- R1 expected = expectedIterator.next();
- R2 actual = actualIterator.next();
- if (!matches(expected, actual, mismatchDescription)) {
- return false;
- }
- }
- return true;
- }
-
- private boolean matches(R1 expected, R2 actual, Description
mismatchDescription) {
- SimpleRecord expectedRecord = convert(expected);
- SimpleRecord actualRecord = convert(actual);
- if (expectedRecord == null) {
- mismatchDescription.appendText("Invalid expected record type: ")
- .appendValue(expected.getClass().getSimpleName());
- return false;
- }
- if (actualRecord == null) {
- mismatchDescription.appendText("Invalid actual record type: ")
- .appendValue(actual.getClass().getSimpleName());
- return false;
- }
- return compare(expectedRecord.key(), actualRecord.key(),
keySerde.deserializer(), "Record key",
- mismatchDescription) &&
- compare(expectedRecord.value(), actualRecord.value(),
valueSerde.deserializer(), "Record value",
- mismatchDescription);
- }
-
- private boolean compare(ByteBuffer lhs,
- ByteBuffer rhs,
- Deserializer<?> deserializer,
- String desc,
- Description mismatchDescription) {
- if ((lhs != null && !lhs.equals(rhs)) || (lhs == null && rhs != null))
{
- mismatchDescription.appendText(desc).appendText(" mismatch.
Expected: ")
-
.appendValue(deserializer.deserialize(topicPartition.topic(),
Utils.toNullableArray(lhs)))
- .appendText("; Actual: ")
-
.appendValue(deserializer.deserialize(topicPartition.topic(),
Utils.toNullableArray(rhs)))
- .appendText("; ");
- return false;
- }
- return true;
- }
-
- @SuppressWarnings("unchecked")
- private SimpleRecord convert(Object recordCandidate) {
- if (recordCandidate instanceof ProducerRecord<?, ?> record) {
- long timestamp = record.timestamp() != null ? record.timestamp() :
RecordBatch.NO_TIMESTAMP;
- ByteBuffer keyBytes =
-
Utils.wrapNullable(keySerde.serializer().serialize(topicPartition.topic(), (K)
record.key()));
- ByteBuffer valueBytes =
-
Utils.wrapNullable(valueSerde.serializer().serialize(topicPartition.topic(),
(V) record.value()));
- Header[] headers = record.headers() != null ?
record.headers().toArray() : Record.EMPTY_HEADERS;
- return new SimpleRecord(timestamp, keyBytes, valueBytes, headers);
- } else if (recordCandidate instanceof ConsumerRecord<?, ?> record) {
- ByteBuffer keyBytes =
-
Utils.wrapNullable(keySerde.serializer().serialize(topicPartition.topic(), (K)
record.key()));
- ByteBuffer valueBytes =
-
Utils.wrapNullable(valueSerde.serializer().serialize(topicPartition.topic(),
(V) record.value()));
- Header[] headers = record.headers() != null ?
record.headers().toArray() : Record.EMPTY_HEADERS;
- return new SimpleRecord(record.timestamp(), keyBytes, valueBytes,
headers);
- } else if (recordCandidate instanceof Record record) {
- return new SimpleRecord(record.timestamp(), record.key(),
record.value(), record.headers());
- } else {
- return null;
- }
- }
-
- /**
- * Provides a matcher which compares the key and value of a sequence of
records with those of
- * the expectedRecords sequence, in order.
- *
- * @param expectedRecords The records expected.
- * @param topicPartition The topic-partition which the records belong to.
- * @param keySerde The {@link Serde} for the keys of the records.
- * @param valueSerde The {@link Serde} for the values of the records.
- */
- public static <R1, R2, K, V> RecordsKeyValueMatcher<R1, R2, K, V>
correspondTo(Collection<R1> expectedRecords,
-
TopicPartition topicPartition,
-
Serde<K> keySerde,
-
Serde<V> valueSerde) {
- return new RecordsKeyValueMatcher<>(expectedRecords, topicPartition,
keySerde, valueSerde);
- }
-}
