You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@kafka.apache.org by gu...@apache.org on 2021/04/18 17:48:16 UTC
[kafka] branch trunk updated: KAFKA-12633: Remove deprecated APIs
in TopologyTestDriver (#10508)
This is an automated email from the ASF dual-hosted git repository.
guozhang 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 035449eb KAFKA-12633: Remove deprecated APIs in TopologyTestDriver (#10508)
035449eb is described below
commit 035449eb5576bf97df888a85edaaf4428b9a19e5
Author: Guozhang Wang <wa...@gmail.com>
AuthorDate: Sun Apr 18 10:46:01 2021 -0700
KAFKA-12633: Remove deprecated APIs in TopologyTestDriver (#10508)
As well as related test classes.
Reviewers: John Roesler <vv...@apache.org>
---
docs/streams/upgrade-guide.html | 1 +
.../apache/kafka/streams/TopologyTestDriver.java | 122 +----
.../kafka/streams/test/ConsumerRecordFactory.java | 583 ---------------------
.../apache/kafka/streams/test/OutputVerifier.java | 458 ----------------
.../kafka/streams/TopologyTestDriverTest.java | 189 +------
.../streams/test/ConsumerRecordFactoryTest.java | 273 ----------
.../kafka/streams/test/OutputVerifierTest.java | 452 ----------------
7 files changed, 25 insertions(+), 2053 deletions(-)
diff --git a/docs/streams/upgrade-guide.html b/docs/streams/upgrade-guide.html
index 0269be9..ea410b1 100644
--- a/docs/streams/upgrade-guide.html
+++ b/docs/streams/upgrade-guide.html
@@ -107,6 +107,7 @@
<li> Overloaded <code>JoinWindows#of, before, after</code>, <code>SessionWindows#with</code>, <code>TimeWindows#of, advanceBy</code>, <code>UnlimitedWindows#startOn</code> and <code>KafkaStreams#close</code> with <code>long</code> typed parameters: deprecated in Kafka 2.1.0 (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-358%3A+Migrate+Streams+API+to+Duration+instead+of+long+ms+times">KIP-358</a>).</li>
<li> Overloaded <code>KStream#groupBy, groupByKey</code> and <code>KTable#groupBy</code> with <code>Serialized</code> parameter: deprecated in Kafka 2.1.0 (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-372%3A+Naming+Repartition+Topics+for+Joins+and+Grouping">KIP-372</a>).</li>
<li> <code>Joined#named, name</code>: deprecated in Kafka 2.3.0 (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-307%3A+Allow+to+define+custom+processor+names+with+KStreams+DSL">KIP-307</a>).</li>
+ <li> <code>TopologyTestDriver#pipeInput, readOutput</code>, <code>OutputVerifier</code> and <code>ConsumerRecordFactory</code> classes (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-470%3A+TopologyTestDriver+test+input+and+output+usability+improvements">KIP-470</a>).</li>
<li> <code>KafkaClientSupplier#getAdminClient</code>: deprecated in Kafka 2.4.0 (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-476%3A+Add+Java+AdminClient+Interface">KIP-476</a>).</li>
<li> Overloaded <code>KStream#join, leftJoin, outerJoin</code> with <code>KStream</code> and <code>Joined</code> parameters: deprecated in Kafka 2.4.0 (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-479%3A+Add+StreamJoined+config+object+to+Join">KIP-479</a>).</li>
<li> Overloaded <code>KafkaStreams#metadataForKey</code>: deprecated in Kafka 2.5.0 (<a href="https://cwiki.apache.org/confluence/display/KAFKA/KIP-535%3A+Allow+state+stores+to+serve+stale+reads+during+rebalance">KIP-535</a>).</li>
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
index bd93367..118bdbb 100644
--- a/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
+++ b/streams/test-utils/src/main/java/org/apache/kafka/streams/TopologyTestDriver.java
@@ -192,7 +192,7 @@ import static org.apache.kafka.streams.state.ValueAndTimestamp.getValueOrNull;
* Note, that calling {@code pipeInput()} will also trigger {@link PunctuationType#STREAM_TIME event-time} base
* {@link ProcessorContext#schedule(Duration, PunctuationType, Punctuator) punctuation} callbacks.
* However, you won't trigger {@link PunctuationType#WALL_CLOCK_TIME wall-clock} type punctuations that you must
- * trigger manually via {@link #advanceWallClockTime(long)}.
+ * trigger manually via {@link #advanceWallClockTime(Duration)}.
* <p>
* Finally, when completed, make sure your tests {@link #close()} the driver to release all resources and
* {@link org.apache.kafka.streams.processor.Processor processors}.
@@ -284,23 +284,6 @@ public class TopologyTestDriver implements Closeable {
this(topology, new Properties(), initialWallClockTimeMs);
}
-
- /**
- * Create a new test diver instance.
- *
- * @deprecated Since 2.4 use {@link #TopologyTestDriver(Topology, Properties, Instant)}
- *
- * @param topology the topology to be tested
- * @param config the configuration for the topology
- * @param initialWallClockTimeMs the initial value of internally mocked wall-clock time
- */
- @Deprecated
- public TopologyTestDriver(final Topology topology,
- final Properties config,
- final long initialWallClockTimeMs) {
- this(topology.internalTopologyBuilder, config, initialWallClockTimeMs);
- }
-
/**
* Create a new test diver instance.
*
@@ -553,25 +536,6 @@ public class TopologyTestDriver implements Closeable {
return Collections.unmodifiableMap(metrics.metrics());
}
- /**
- * Send an input message with the given key, value, and timestamp on the specified topic to the topology and then
- * commit the messages.
- *
- * @deprecated Since 2.4 use methods of {@link TestInputTopic} instead
- *
- * @param consumerRecord the record to be processed
- */
- @Deprecated
- public void pipeInput(final ConsumerRecord<byte[], byte[]> consumerRecord) {
- pipeRecord(
- consumerRecord.topic(),
- consumerRecord.timestamp(),
- consumerRecord.key(),
- consumerRecord.value(),
- consumerRecord.headers()
- );
- }
-
private void pipeRecord(final String topicName,
final long timestamp,
final byte[] key,
@@ -738,34 +702,6 @@ public class TopologyTestDriver implements Closeable {
}
/**
- * Send input messages to the topology and then commit each message individually.
- *
- * @deprecated Since 2.4 use methods of {@link TestInputTopic} instead
- *
- * @param records a list of records to be processed
- */
- @Deprecated
- public void pipeInput(final List<ConsumerRecord<byte[], byte[]>> records) {
- for (final ConsumerRecord<byte[], byte[]> record : records) {
- pipeInput(record);
- }
- }
-
- /**
- * Advances the internally mocked wall-clock time.
- * This might trigger a {@link PunctuationType#WALL_CLOCK_TIME wall-clock} type
- * {@link ProcessorContext#schedule(Duration, PunctuationType, Punctuator) punctuations}.
- *
- * @deprecated Since 2.4 use {@link #advanceWallClockTime(Duration)} instead
- *
- * @param advanceMs the amount of time to advance wall-clock time in milliseconds
- */
- @Deprecated
- public void advanceWallClockTime(final long advanceMs) {
- advanceWallClockTime(Duration.ofMillis(advanceMs));
- }
-
- /**
* Advances the internally mocked wall-clock time.
* This might trigger a {@link PunctuationType#WALL_CLOCK_TIME wall-clock} type
* {@link ProcessorContext#schedule(Duration, PunctuationType, Punctuator) punctuations}.
@@ -783,48 +719,6 @@ public class TopologyTestDriver implements Closeable {
completeAllProcessableWork();
}
- /**
- * Read the next record from the given topic.
- * These records were output by the topology during the previous calls to {@link #pipeInput(ConsumerRecord)}.
- *
- * @deprecated Since 2.4 use methods of {@link TestOutputTopic} instead
- *
- * @param topic the name of the topic
- * @return the next record on that topic, or {@code null} if there is no record available
- */
- @Deprecated
- public ProducerRecord<byte[], byte[]> readOutput(final String topic) {
- final Queue<ProducerRecord<byte[], byte[]>> outputRecords = outputRecordsByTopic.get(topic);
- if (outputRecords == null) {
- return null;
- }
- return outputRecords.poll();
- }
-
- /**
- * Read the next record from the given topic.
- * These records were output by the topology during the previous calls to {@link #pipeInput(ConsumerRecord)}.
- *
- * @deprecated Since 2.4 use methods of {@link TestOutputTopic} instead
- *
- * @param topic the name of the topic
- * @param keyDeserializer the deserializer for the key type
- * @param valueDeserializer the deserializer for the value type
- * @return the next record on that topic, or {@code null} if there is no record available
- */
- @Deprecated
- public <K, V> ProducerRecord<K, V> readOutput(final String topic,
- final Deserializer<K> keyDeserializer,
- final Deserializer<V> valueDeserializer) {
- final ProducerRecord<byte[], byte[]> record = readOutput(topic);
- if (record == null) {
- return null;
- }
- final K key = keyDeserializer.deserialize(record.topic(), record.headers(), record.key());
- final V value = valueDeserializer.deserialize(record.topic(), record.headers(), record.value());
- return new ProducerRecord<>(record.topic(), record.partition(), record.timestamp(), key, value, record.headers());
- }
-
private Queue<ProducerRecord<byte[], byte[]>> getRecordsQueue(final String topicName) {
final Queue<ProducerRecord<byte[], byte[]>> outputRecords = outputRecordsByTopic.get(topicName);
if (outputRecords == null && !processorTopology.sinkTopics().contains(topicName)) {
@@ -966,7 +860,7 @@ public class TopologyTestDriver implements Closeable {
* The stores can be a "regular" or global stores.
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord)} process an input message}, and/or to check the store afterward.
* <p>
* Note, that {@code StateStore} might be {@code null} if a store is added but not connected to any processor.
* <p>
@@ -998,7 +892,7 @@ public class TopologyTestDriver implements Closeable {
* For built-in stores, the corresponding typed methods like {@link #getKeyValueStore(String)} should be used.
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord) process an input message}, and/or to check the store afterward.
*
* @param name the name of the store
* @return the state store, or {@code null} if no store has been registered with the given name
@@ -1074,7 +968,7 @@ public class TopologyTestDriver implements Closeable {
* {@link #getTimestampedKeyValueStore(String)} for full store access instead.</strong>
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord) process an input message}, and/or to check the store afterward.
*
* @param name the name of the store
* @return the key value store, or {@code null} if no {@link KeyValueStore} or {@link TimestampedKeyValueStore}
@@ -1101,7 +995,7 @@ public class TopologyTestDriver implements Closeable {
* The store can be a "regular" or global store.
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord) process an input message}, and/or to check the store afterward.
*
* @param name the name of the store
* @return the key value store, or {@code null} if no {@link TimestampedKeyValueStore} has been registered with the given name
@@ -1127,7 +1021,7 @@ public class TopologyTestDriver implements Closeable {
* {@link #getTimestampedWindowStore(String)} for full store access instead.</strong>
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord) process an input message}, and/or to check the store afterward.
*
* @param name the name of the store
* @return the key value store, or {@code null} if no {@link WindowStore} or {@link TimestampedWindowStore}
@@ -1154,7 +1048,7 @@ public class TopologyTestDriver implements Closeable {
* The store can be a "regular" or global store.
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord) process an input message}, and/or to check the store afterward.
*
* @param name the name of the store
* @return the key value store, or {@code null} if no {@link TimestampedWindowStore} has been registered with the given name
@@ -1176,7 +1070,7 @@ public class TopologyTestDriver implements Closeable {
* The store can be a "regular" or global store.
* <p>
* This is often useful in test cases to pre-populate the store before the test case instructs the topology to
- * {@link #pipeInput(ConsumerRecord) process an input message}, and/or to check the store afterward.
+ * {@link TestInputTopic#pipeInput(TestRecord) process an input message}, and/or to check the store afterward.
*
* @param name the name of the store
* @return the key value store, or {@code null} if no {@link SessionStore} has been registered with the given name
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/test/ConsumerRecordFactory.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/test/ConsumerRecordFactory.java
deleted file mode 100644
index 15164f6..0000000
--- a/streams/test-utils/src/main/java/org/apache/kafka/streams/test/ConsumerRecordFactory.java
+++ /dev/null
@@ -1,583 +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.streams.test;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.header.internals.RecordHeaders;
-import org.apache.kafka.common.record.TimestampType;
-import org.apache.kafka.common.serialization.Serializer;
-import org.apache.kafka.streams.KeyValue;
-import org.apache.kafka.streams.TestInputTopic;
-import org.apache.kafka.streams.TopologyTestDriver;
-
-import java.util.ArrayList;
-import java.util.List;
-import java.util.Objects;
-import java.util.Optional;
-
-/**
- * Factory to create {@link ConsumerRecord consumer records} for a single single-partitioned topic with given key and
- * value {@link Serializer serializers}.
- *
- * @deprecated Since 2.4 use methods of {@link TestInputTopic} instead
- *
- * @param <K> the type of the key
- * @param <V> the type of the value
- *
- * @see TopologyTestDriver
- */
-@Deprecated
-public class ConsumerRecordFactory<K, V> {
- private final String topicName;
- private final Serializer<K> keySerializer;
- private final Serializer<V> valueSerializer;
- private long timeMs;
- private final long advanceMs;
-
- /**
- * Create a new factory for the given topic.
- * Uses current system time as start timestamp.
- * Auto-advance is disabled.
- *
- * @param keySerializer the key serializer
- * @param valueSerializer the value serializer
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecordFactory(final Serializer<K> keySerializer,
- final Serializer<V> valueSerializer) {
- this(null, keySerializer, valueSerializer, System.currentTimeMillis());
- }
-
- /**
- * Create a new factory for the given topic.
- * Uses current system time as start timestamp.
- * Auto-advance is disabled.
- *
- * @param defaultTopicName the default topic name used for all generated {@link ConsumerRecord consumer records}
- * @param keySerializer the key serializer
- * @param valueSerializer the value serializer
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecordFactory(final String defaultTopicName,
- final Serializer<K> keySerializer,
- final Serializer<V> valueSerializer) {
- this(defaultTopicName, keySerializer, valueSerializer, System.currentTimeMillis());
- }
-
- /**
- * Create a new factory for the given topic.
- * Auto-advance is disabled.
- *
- * @param keySerializer the key serializer
- * @param valueSerializer the value serializer
- * @param startTimestampMs the initial timestamp for generated records
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecordFactory(final Serializer<K> keySerializer,
- final Serializer<V> valueSerializer,
- final long startTimestampMs) {
- this(null, keySerializer, valueSerializer, startTimestampMs, 0L);
- }
-
- /**
- * Create a new factory for the given topic.
- * Auto-advance is disabled.
- *
- * @param defaultTopicName the topic name used for all generated {@link ConsumerRecord consumer records}
- * @param keySerializer the key serializer
- * @param valueSerializer the value serializer
- * @param startTimestampMs the initial timestamp for generated records
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecordFactory(final String defaultTopicName,
- final Serializer<K> keySerializer,
- final Serializer<V> valueSerializer,
- final long startTimestampMs) {
- this(defaultTopicName, keySerializer, valueSerializer, startTimestampMs, 0L);
- }
-
- /**
- * Create a new factory for the given topic.
- *
- * @param keySerializer the key serializer
- * @param valueSerializer the value serializer
- * @param startTimestampMs the initial timestamp for generated records
- * @param autoAdvanceMs the time increment pre generated record
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecordFactory(final Serializer<K> keySerializer,
- final Serializer<V> valueSerializer,
- final long startTimestampMs,
- final long autoAdvanceMs) {
- this(null, keySerializer, valueSerializer, startTimestampMs, autoAdvanceMs);
- }
-
- /**
- * Create a new factory for the given topic.
- *
- * @param defaultTopicName the topic name used for all generated {@link ConsumerRecord consumer records}
- * @param keySerializer the key serializer
- * @param valueSerializer the value serializer
- * @param startTimestampMs the initial timestamp for generated records
- * @param autoAdvanceMs the time increment pre generated record
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecordFactory(final String defaultTopicName,
- final Serializer<K> keySerializer,
- final Serializer<V> valueSerializer,
- final long startTimestampMs,
- final long autoAdvanceMs) {
- Objects.requireNonNull(keySerializer, "keySerializer cannot be null");
- Objects.requireNonNull(valueSerializer, "valueSerializer cannot be null");
- this.topicName = defaultTopicName;
- this.keySerializer = keySerializer;
- this.valueSerializer = valueSerializer;
- timeMs = startTimestampMs;
- advanceMs = autoAdvanceMs;
- }
-
- /**
- * Advances the internally tracked time.
- *
- * @param advanceMs the amount of time to advance
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public void advanceTimeMs(final long advanceMs) {
- if (advanceMs < 0) {
- throw new IllegalArgumentException("advanceMs must be positive");
- }
- timeMs += advanceMs;
- }
-
- /**
- * Create a {@link ConsumerRecord} with the given topic name, key, value, headers, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param topicName the topic name
- * @param key the record key
- * @param value the record value
- * @param headers the record headers
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final K key,
- final V value,
- final Headers headers,
- final long timestampMs) {
- Objects.requireNonNull(topicName, "topicName cannot be null.");
- Objects.requireNonNull(headers, "headers cannot be null.");
- final byte[] serializedKey = keySerializer.serialize(topicName, headers, key);
- final byte[] serializedValue = valueSerializer.serialize(topicName, headers, value);
- return new ConsumerRecord<>(
- topicName,
- -1,
- -1L,
- timestampMs,
- TimestampType.CREATE_TIME,
- serializedKey == null ? 0 : serializedKey.length,
- serializedValue == null ? 0 : serializedValue.length,
- serializedKey,
- serializedValue,
- headers,
- Optional.empty());
- }
-
- /**
- * Create a {@link ConsumerRecord} with the given topic name and given topic, key, value, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param topicName the topic name
- * @param key the record key
- * @param value the record value
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final K key,
- final V value,
- final long timestampMs) {
- return create(topicName, key, value, new RecordHeaders(), timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and given key, value, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param key the record key
- * @param value the record value
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final K key,
- final V value,
- final long timestampMs) {
- return create(key, value, new RecordHeaders(), timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and given key, value, headers, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param key the record key
- * @param value the record value
- * @param headers the record headers
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final K key,
- final V value,
- final Headers headers,
- final long timestampMs) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, K key, V value, long timestampMs) instead.");
- }
- return create(topicName, key, value, headers, timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with the given topic name, key, and value.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param topicName the topic name
- * @param key the record key
- * @param value the record value
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final K key,
- final V value) {
- final long timestamp = timeMs;
- timeMs += advanceMs;
- return create(topicName, key, value, new RecordHeaders(), timestamp);
- }
-
- /**
- * Create a {@link ConsumerRecord} with the given topic name, key, value, and headers.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param topicName the topic name
- * @param key the record key
- * @param value the record value
- * @param headers the record headers
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final K key,
- final V value,
- final Headers headers) {
- final long timestamp = timeMs;
- timeMs += advanceMs;
- return create(topicName, key, value, headers, timestamp);
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and given key and value.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param key the record key
- * @param value the record value
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final K key,
- final V value) {
- return create(key, value, new RecordHeaders());
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and given key, value, and headers.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param key the record key
- * @param value the record value
- * @param headers the record headers
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final K key,
- final V value,
- final Headers headers) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, K key, V value) instead.");
- }
- return create(topicName, key, value, headers);
- }
-
- /**
- * Create a {@link ConsumerRecord} with {@code null}-key and the given topic name, value, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param topicName the topic name
- * @param value the record value
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final V value,
- final long timestampMs) {
- return create(topicName, null, value, new RecordHeaders(), timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with {@code null}-key and the given topic name, value, headers, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param topicName the topic name
- * @param value the record value
- * @param headers the record headers
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final V value,
- final Headers headers,
- final long timestampMs) {
- return create(topicName, null, value, headers, timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and {@code null}-key as well as given value and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param value the record value
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final V value,
- final long timestampMs) {
- return create(value, new RecordHeaders(), timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and {@code null}-key as well as given value, headers, and timestamp.
- * Does not auto advance internally tracked time.
- *
- * @param value the record value
- * @param headers the record headers
- * @param timestampMs the record timestamp
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final V value,
- final Headers headers,
- final long timestampMs) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, V value, long timestampMs) instead.");
- }
- return create(topicName, value, headers, timestampMs);
- }
-
- /**
- * Create a {@link ConsumerRecord} with {@code null}-key and the given topic name, value, and headers.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param topicName the topic name
- * @param value the record value
- * @param headers the record headers
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final V value,
- final Headers headers) {
- return create(topicName, null, value, headers);
- }
-
- /**
- * Create a {@link ConsumerRecord} with {@code null}-key and the given topic name and value.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param topicName the topic name
- * @param value the record value
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final String topicName,
- final V value) {
- return create(topicName, null, value, new RecordHeaders());
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and {@code null}-key was well as given value.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param value the record value
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final V value) {
- return create(value, new RecordHeaders());
- }
-
- /**
- * Create a {@link ConsumerRecord} with default topic name and {@code null}-key was well as given value and headers.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param value the record value
- * @param headers the record headers
- * @return the generated {@link ConsumerRecord}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public ConsumerRecord<byte[], byte[]> create(final V value,
- final Headers headers) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, V value, long timestampMs) instead.");
- }
- return create(topicName, value, headers);
- }
-
- /**
- * Creates {@link ConsumerRecord consumer records} with the given topic name, keys, and values.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param topicName the topic name
- * @param keyValues the record keys and values
- * @return the generated {@link ConsumerRecord consumer records}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public List<ConsumerRecord<byte[], byte[]>> create(final String topicName,
- final List<KeyValue<K, V>> keyValues) {
- final List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>(keyValues.size());
-
- for (final KeyValue<K, V> keyValue : keyValues) {
- records.add(create(topicName, keyValue.key, keyValue.value));
- }
-
- return records;
- }
-
- /**
- * Creates {@link ConsumerRecord consumer records} with default topic name as well as given keys and values.
- * The timestamp will be generated based on the constructor provided start time and time will auto advance.
- *
- * @param keyValues the record keys and values
- * @return the generated {@link ConsumerRecord consumer records}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public List<ConsumerRecord<byte[], byte[]>> create(final List<KeyValue<K, V>> keyValues) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, List<KeyValue<K, V>> keyValues) instead.");
- }
-
- return create(topicName, keyValues);
- }
-
- /**
- * Creates {@link ConsumerRecord consumer records} with the given topic name, keys, and values.
- * Does not auto advance internally tracked time.
- *
- * @param topicName the topic name
- * @param keyValues the record keys and values
- * @param startTimestamp the timestamp for the first generated record
- * @param advanceMs the time difference between two consecutive generated records
- * @return the generated {@link ConsumerRecord consumer records}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public List<ConsumerRecord<byte[], byte[]>> create(final String topicName,
- final List<KeyValue<K, V>> keyValues,
- final long startTimestamp,
- final long advanceMs) {
- if (advanceMs < 0) {
- throw new IllegalArgumentException("advanceMs must be positive");
- }
-
- final List<ConsumerRecord<byte[], byte[]>> records = new ArrayList<>(keyValues.size());
-
- long timestamp = startTimestamp;
- for (final KeyValue<K, V> keyValue : keyValues) {
- records.add(create(topicName, keyValue.key, keyValue.value, new RecordHeaders(), timestamp));
- timestamp += advanceMs;
- }
-
- return records;
- }
-
- /**
- * Creates {@link ConsumerRecord consumer records} with default topic name as well as given keys and values.
- * Does not auto advance internally tracked time.
- *
- * @param keyValues the record keys and values
- * @param startTimestamp the timestamp for the first generated record
- * @param advanceMs the time difference between two consecutive generated records
- * @return the generated {@link ConsumerRecord consumer records}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public List<ConsumerRecord<byte[], byte[]>> create(final List<KeyValue<K, V>> keyValues,
- final long startTimestamp,
- final long advanceMs) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, List<KeyValue<K, V>> keyValues, long startTimestamp, long advanceMs) instead.");
- }
-
- return create(topicName, keyValues, startTimestamp, advanceMs);
- }
-
- /**
- * Creates {@link ConsumerRecord consumer records} with the given topic name, keys and values.
- * For each generated record, the time is advanced by 1.
- * Does not auto advance internally tracked time.
- *
- * @param topicName the topic name
- * @param keyValues the record keys and values
- * @param startTimestamp the timestamp for the first generated record
- * @return the generated {@link ConsumerRecord consumer records}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public List<ConsumerRecord<byte[], byte[]>> create(final String topicName,
- final List<KeyValue<K, V>> keyValues,
- final long startTimestamp) {
- return create(topicName, keyValues, startTimestamp, 1);
- }
-
- /**
- * Creates {@link ConsumerRecord consumer records} with the given keys and values.
- * For each generated record, the time is advanced by 1.
- * Does not auto advance internally tracked time.
- *
- * @param keyValues the record keys and values
- * @param startTimestamp the timestamp for the first generated record
- * @return the generated {@link ConsumerRecord consumer records}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public List<ConsumerRecord<byte[], byte[]>> create(final List<KeyValue<K, V>> keyValues,
- final long startTimestamp) {
- if (topicName == null) {
- throw new IllegalStateException("ConsumerRecordFactory was created without defaultTopicName. " +
- "Use #create(String topicName, List<KeyValue<K, V>> keyValues, long startTimestamp) instead.");
- }
-
- return create(topicName, keyValues, startTimestamp, 1);
- }
-
-}
diff --git a/streams/test-utils/src/main/java/org/apache/kafka/streams/test/OutputVerifier.java b/streams/test-utils/src/main/java/org/apache/kafka/streams/test/OutputVerifier.java
deleted file mode 100644
index af0f51c..0000000
--- a/streams/test-utils/src/main/java/org/apache/kafka/streams/test/OutputVerifier.java
+++ /dev/null
@@ -1,458 +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.streams.test;
-
-import org.apache.kafka.clients.producer.ProducerRecord;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.streams.TestOutputTopic;
-import org.apache.kafka.streams.TopologyTestDriver;
-
-import java.util.Objects;
-
-/**
- * Helper class to verify topology result records.
- *
- * @deprecated Since 2.4 use methods of {@link TestOutputTopic} and standard assertion libraries instead
- *
- * @see TopologyTestDriver
- */
-@Deprecated
-public class OutputVerifier {
-
- /**
- * Compares a {@link ProducerRecord} with the provided value and throws an {@link AssertionError} if the
- * {@code ProducerRecord}'s value is not equal to the expected value.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s value is not equal to {@code expectedValue}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareValue(final ProducerRecord<K, V> record,
- final V expectedValue) throws AssertionError {
- Objects.requireNonNull(record);
-
- final V recordValue = record.value();
- final AssertionError error = new AssertionError("Expected value=" + expectedValue + " but was value=" + recordValue);
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
- }
-
- /**
- * Compares the values of two {@link ProducerRecord}'s and throws an {@link AssertionError} if they are not equal to
- * each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s value is not equal to {@code expectedRecord}'s value
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareValue(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareValue(record, expectedRecord.value());
- }
-
- /**
- * Compares a {@link ProducerRecord} with the provided key and value and throws an {@link AssertionError} if the
- * {@code ProducerRecord}'s key or value is not equal to the expected key or value.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedKey the expected key of the {@code ProducerRecord}
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key or value is not equal to {@code expectedKey} or {@code expectedValue}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValue(final ProducerRecord<K, V> record,
- final K expectedKey,
- final V expectedValue) throws AssertionError {
- Objects.requireNonNull(record);
-
- final K recordKey = record.key();
- final V recordValue = record.value();
- final AssertionError error = new AssertionError("Expected <" + expectedKey + ", " + expectedValue + "> " +
- "but was <" + recordKey + ", " + recordValue + ">");
-
- if (recordKey != null) {
- if (!recordKey.equals(expectedKey)) {
- throw error;
- }
- } else if (expectedKey != null) {
- throw error;
- }
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
- }
-
- /**
- * Compares the keys and values of two {@link ProducerRecord}'s and throws an {@link AssertionError} if the keys or
- * values are not equal to each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key or value is not equal to {@code expectedRecord}'s key or value
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValue(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareKeyValue(record, expectedRecord.key(), expectedRecord.value());
- }
-
- /**
- * Compares a {@link ProducerRecord} with the provided value and timestamp and throws an {@link AssertionError} if
- * the {@code ProducerRecord}'s value or timestamp is not equal to the expected value or timestamp.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param expectedTimestamp the expected timestamps of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s value or timestamp is not equal to {@code expectedValue} or {@code expectedTimestamp}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareValueTimestamp(final ProducerRecord<K, V> record,
- final V expectedValue,
- final long expectedTimestamp) throws AssertionError {
- Objects.requireNonNull(record);
-
- final V recordValue = record.value();
- final long recordTimestamp = record.timestamp();
- final AssertionError error = new AssertionError("Expected value=" + expectedValue + " with timestamp=" + expectedTimestamp +
- " but was value=" + recordValue + " with timestamp=" + recordTimestamp);
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
-
- if (recordTimestamp != expectedTimestamp) {
- throw error;
- }
- }
-
- /**
- * Compares the values and timestamps of two {@link ProducerRecord}'s and throws an {@link AssertionError} if the
- * values or timestamps are not equal to each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s value or timestamp is not equal to {@code expectedRecord}'s value or timestamp
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareValueTimestamp(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareValueTimestamp(record, expectedRecord.value(), expectedRecord.timestamp());
- }
-
- /**
- * Compares a {@link ProducerRecord} with the provided key, value, and timestamp and throws an
- * {@link AssertionError} if the {@code ProducerRecord}'s key, value, or timestamp is not equal to the expected key,
- * value, or timestamp.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedKey the expected key of the {@code ProducerRecord}
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param expectedTimestamp the expected timestamp of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key, value, timestamp is not equal to {@code expectedKey},
- * {@code expectedValue}, or {@code expectedTimestamps}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValueTimestamp(final ProducerRecord<K, V> record,
- final K expectedKey,
- final V expectedValue,
- final long expectedTimestamp) throws AssertionError {
- Objects.requireNonNull(record);
-
- final K recordKey = record.key();
- final V recordValue = record.value();
- final long recordTimestamp = record.timestamp();
- final AssertionError error = new AssertionError("Expected <" + expectedKey + ", " + expectedValue + "> with timestamp=" + expectedTimestamp +
- " but was <" + recordKey + ", " + recordValue + "> with timestamp=" + recordTimestamp);
-
- if (recordKey != null) {
- if (!recordKey.equals(expectedKey)) {
- throw error;
- }
- } else if (expectedKey != null) {
- throw error;
- }
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
-
- if (recordTimestamp != expectedTimestamp) {
- throw error;
- }
- }
-
- /**
- * Compares the keys, values, and timestamps of two {@link ProducerRecord}'s and throws an {@link AssertionError} if
- * the keys, values, or timestamps are not equal to each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key, value, or timestamp is not equal to
- * {@code expectedRecord}'s key, value, or timestamp
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValueTimestamp(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareKeyValueTimestamp(record, expectedRecord.key(), expectedRecord.value(), expectedRecord.timestamp());
- }
-
- /**
- * Compares a {@link ProducerRecord} with the provided value and headers and throws an {@link AssertionError} if
- * the {@code ProducerRecord}'s value or headers is not equal to the expected value or headers.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param expectedHeaders the expected headers of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s value or headers is not equal to {@code expectedValue} or {@code expectedHeaders}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareValueHeaders(final ProducerRecord<K, V> record,
- final V expectedValue,
- final Headers expectedHeaders) throws AssertionError {
- Objects.requireNonNull(record);
-
- final V recordValue = record.value();
- final Headers recordHeaders = record.headers();
- final AssertionError error = new AssertionError("Expected value=" + expectedValue + " with headers=" + expectedHeaders +
- " but was value=" + recordValue + " with headers=" + recordHeaders);
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
-
- if (recordHeaders != null) {
- if (!recordHeaders.equals(expectedHeaders)) {
- throw error;
- }
- } else if (expectedHeaders != null) {
- throw error;
- }
- }
-
- /**
- * Compares the values and headers of two {@link ProducerRecord}'s and throws an {@link AssertionError} if the
- * values or headers are not equal to each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s value or headers is not equal to {@code expectedRecord}'s value or headers
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareValueHeaders(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareValueHeaders(record, expectedRecord.value(), expectedRecord.headers());
- }
-
- /**
- * Compares a {@link ProducerRecord} with the provided key, value, and headers and throws an
- * {@link AssertionError} if the {@code ProducerRecord}'s key, value, or headers is not equal to the expected key,
- * value, or headers.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedKey the expected key of the {@code ProducerRecord}
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param expectedHeaders the expected headers of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key, value, headers is not equal to {@code expectedKey},
- * {@code expectedValue}, or {@code expectedHeaders}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValueHeaders(final ProducerRecord<K, V> record,
- final K expectedKey,
- final V expectedValue,
- final Headers expectedHeaders) throws AssertionError {
- Objects.requireNonNull(record);
-
- final K recordKey = record.key();
- final V recordValue = record.value();
- final Headers recordHeaders = record.headers();
- final AssertionError error = new AssertionError("Expected <" + expectedKey + ", " + expectedValue + "> with headers=" + expectedHeaders +
- " but was <" + recordKey + ", " + recordValue + "> with headers=" + recordHeaders);
-
- if (recordKey != null) {
- if (!recordKey.equals(expectedKey)) {
- throw error;
- }
- } else if (expectedKey != null) {
- throw error;
- }
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
-
- if (recordHeaders != null) {
- if (!recordHeaders.equals(expectedHeaders)) {
- throw error;
- }
- } else if (expectedHeaders != null) {
- throw error;
- }
- }
-
- /**
- * Compares the keys, values, and headers of two {@link ProducerRecord}'s and throws an {@link AssertionError} if
- * the keys, values, or headers are not equal to each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key, value, or headers is not equal to
- * {@code expectedRecord}'s key, value, or headers
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValueHeaders(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareKeyValueHeaders(record, expectedRecord.key(), expectedRecord.value(), expectedRecord.headers());
- }
-
- /**
- * Compares a {@link ProducerRecord} with the provided key, value, headers, and timestamp and throws an
- * {@link AssertionError} if the {@code ProducerRecord}'s key, value, headers, or timestamp is not equal to the expected key,
- * value, headers, or timestamp.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedKey the expected key of the {@code ProducerRecord}
- * @param expectedValue the expected value of the {@code ProducerRecord}
- * @param expectedHeaders the expected headers of the {@code ProducerRecord}
- * @param expectedTimestamp the expected timestamp of the {@code ProducerRecord}
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key, value, headers is not equal to {@code expectedKey},
- * {@code expectedValue}, or {@code expectedHeaders}
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValueHeadersTimestamp(final ProducerRecord<K, V> record,
- final K expectedKey,
- final V expectedValue,
- final Headers expectedHeaders,
- final long expectedTimestamp) throws AssertionError {
- Objects.requireNonNull(record);
-
- final K recordKey = record.key();
- final V recordValue = record.value();
- final Headers recordHeaders = record.headers();
- final long recordTimestamp = record.timestamp();
- final AssertionError error = new AssertionError("Expected <" + expectedKey + ", " + expectedValue + ">" +
- " with timestamp=" + expectedTimestamp + " and headers=" + expectedHeaders +
- " but was <" + recordKey + ", " + recordValue + ">" +
- " with timestamp=" + recordTimestamp + " and headers=" + recordHeaders);
-
- if (recordKey != null) {
- if (!recordKey.equals(expectedKey)) {
- throw error;
- }
- } else if (expectedKey != null) {
- throw error;
- }
-
- if (recordValue != null) {
- if (!recordValue.equals(expectedValue)) {
- throw error;
- }
- } else if (expectedValue != null) {
- throw error;
- }
-
- if (recordHeaders != null) {
- if (!recordHeaders.equals(expectedHeaders)) {
- throw error;
- }
- } else if (expectedHeaders != null) {
- throw error;
- }
-
- if (recordTimestamp != expectedTimestamp) {
- throw error;
- }
- }
-
- /**
- * Compares the keys, values, headers, and timestamp of two {@link ProducerRecord}'s and throws an {@link AssertionError} if
- * the keys, values, headers, or timestamps are not equal to each other.
- *
- * @param record a output {@code ProducerRecord} for verification
- * @param expectedRecord a {@code ProducerRecord} for verification
- * @param <K> the key type
- * @param <V> the value type
- * @throws AssertionError if {@code ProducerRecord}'s key, value, headers, or timestamp is not equal to
- * {@code expectedRecord}'s key, value, headers, or timestamp
- */
- @SuppressWarnings({"WeakerAccess", "unused"})
- public static <K, V> void compareKeyValueHeadersTimestamp(final ProducerRecord<K, V> record,
- final ProducerRecord<K, V> expectedRecord) throws AssertionError {
- Objects.requireNonNull(expectedRecord);
- compareKeyValueHeadersTimestamp(record, expectedRecord.key(), expectedRecord.value(), expectedRecord.headers(), expectedRecord.timestamp());
- }
-}
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
index d31dc53..ad7bc16 100644
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
+++ b/streams/test-utils/src/test/java/org/apache/kafka/streams/TopologyTestDriverTest.java
@@ -16,7 +16,6 @@
*/
package org.apache.kafka.streams;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
@@ -90,7 +89,6 @@ import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertThrows;
import static org.junit.jupiter.api.Assertions.assertTrue;
-import static org.junit.jupiter.api.Assertions.fail;
public abstract class TopologyTestDriverTest {
@@ -118,17 +116,6 @@ public abstract class TopologyTestDriverTest {
private final byte[] value2 = new byte[0];
private final long timestamp2 = 43L;
- //Factory and records for testing already deprecated methods
- @SuppressWarnings("deprecation")
- private final org.apache.kafka.streams.test.ConsumerRecordFactory<byte[], byte[]> consumerRecordFactory =
- new org.apache.kafka.streams.test.ConsumerRecordFactory<>(
- new ByteArraySerializer(),
- new ByteArraySerializer());
- private final ConsumerRecord<byte[], byte[]> consumerRecord1 =
- consumerRecordFactory.create(SOURCE_TOPIC_1, key1, value1, headers, timestamp1);
- private final ConsumerRecord<byte[], byte[]> consumerRecord2 =
- consumerRecordFactory.create(SOURCE_TOPIC_2, key2, value2, timestamp2);
-
private TopologyTestDriver testDriver;
private final Properties config;
private KeyValueStore<String, Long> store;
@@ -144,16 +131,6 @@ public abstract class TopologyTestDriverTest {
private final String topic;
private final Headers headers;
- TTDTestRecord(final ConsumerRecord<byte[], byte[]> consumerRecord,
- final long newOffset) {
- key = consumerRecord.key();
- value = consumerRecord.value();
- timestamp = consumerRecord.timestamp();
- offset = newOffset;
- topic = consumerRecord.topic();
- headers = consumerRecord.headers();
- }
-
TTDTestRecord(final String newTopic,
final TestRecord<byte[], byte[]> consumerRecord,
final long newOffset) {
@@ -181,7 +158,12 @@ public abstract class TopologyTestDriverTest {
@Override
public String toString() {
- return "key: " + key + ", value: " + value + ", timestamp: " + timestamp + ", offset: " + offset + ", topic: " + topic;
+ return "key: " + key +
+ ", value: " + value +
+ ", timestamp: " + timestamp +
+ ", offset: " + offset +
+ ", topic: " + topic +
+ ", num.headers: " + (headers == null ? "null" : headers.toArray().length);
}
@Override
@@ -481,25 +463,6 @@ public abstract class TopologyTestDriverTest {
null));
}
- @Deprecated
- @Test
- public void shouldThrowForUnknownTopicDeprecated() {
- final String unknownTopic = "unknownTopic";
- final org.apache.kafka.streams.test.ConsumerRecordFactory<byte[], byte[]> consumerRecordFactory =
- new org.apache.kafka.streams.test.ConsumerRecordFactory<>(
- "unknownTopic",
- new ByteArraySerializer(),
- new ByteArraySerializer());
-
- testDriver = new TopologyTestDriver(new Topology());
- try {
- testDriver.pipeInput(consumerRecordFactory.create((byte[]) null));
- fail("Should have throw IllegalArgumentException");
- } catch (final IllegalArgumentException exception) {
- assertEquals("Unknown topic: " + unknownTopic, exception.getMessage());
- }
- }
-
@Test
public void shouldThrowNoSuchElementExceptionForUnusedOutputTopicWithDynamicRouting() {
testDriver = new TopologyTestDriver(setupSourceSinkTopology());
@@ -605,97 +568,37 @@ public abstract class TopologyTestDriverTest {
}
- @Deprecated
- //Test not migrated to non-deprecated methods, topic handling not based on record any more
@Test
- public void shouldSendRecordViaCorrectSourceTopicDeprecated() {
+ public void shouldSendRecordViaCorrectSourceTopic() {
testDriver = new TopologyTestDriver(setupMultipleSourceTopology(SOURCE_TOPIC_1, SOURCE_TOPIC_2));
final List<TTDTestRecord> processedRecords1 = mockProcessors.get(0).processedRecords;
final List<TTDTestRecord> processedRecords2 = mockProcessors.get(1).processedRecords;
- testDriver.pipeInput(consumerRecord1);
+ final TestInputTopic<byte[], byte[]> inputTopic1 = testDriver.createInputTopic(SOURCE_TOPIC_1,
+ new ByteArraySerializer(), new ByteArraySerializer());
+ final TestInputTopic<byte[], byte[]> inputTopic2 = testDriver.createInputTopic(SOURCE_TOPIC_2,
+ new ByteArraySerializer(), new ByteArraySerializer());
+
+ inputTopic1.pipeInput(new TestRecord<>(key1, value1, headers, timestamp1));
assertEquals(1, processedRecords1.size());
assertEquals(0, processedRecords2.size());
TTDTestRecord record = processedRecords1.get(0);
- TTDTestRecord expectedResult = new TTDTestRecord(consumerRecord1, 0L);
+ TTDTestRecord expectedResult = new TTDTestRecord(key1, value1, headers, timestamp1, 0L, SOURCE_TOPIC_1);
assertThat(record, equalTo(expectedResult));
- testDriver.pipeInput(consumerRecord2);
+ inputTopic2.pipeInput(new TestRecord<>(key2, value2, Instant.ofEpochMilli(timestamp2)));
assertEquals(1, processedRecords1.size());
assertEquals(1, processedRecords2.size());
record = processedRecords2.get(0);
- expectedResult = new TTDTestRecord(consumerRecord2, 0L);
+ expectedResult = new TTDTestRecord(key2, value2, new RecordHeaders((Iterable<Header>) null), timestamp2, 0L, SOURCE_TOPIC_2);
assertThat(record, equalTo(expectedResult));
}
- @Deprecated
- @Test
- public void shouldUseSourceSpecificDeserializersDeprecated() {
- final Topology topology = new Topology();
-
- final String sourceName1 = "source-1";
- final String sourceName2 = "source-2";
- final String processor = "processor";
-
- topology.addSource(sourceName1, Serdes.Long().deserializer(), Serdes.String().deserializer(), SOURCE_TOPIC_1);
- topology.addSource(sourceName2, Serdes.Integer().deserializer(), Serdes.Double().deserializer(), SOURCE_TOPIC_2);
- topology.addProcessor(processor, new MockProcessorSupplier(), sourceName1, sourceName2);
- topology.addSink(
- "sink",
- SINK_TOPIC_1,
- (topic, data) -> {
- if (data instanceof Long) {
- return Serdes.Long().serializer().serialize(topic, (Long) data);
- }
- return Serdes.Integer().serializer().serialize(topic, (Integer) data);
- },
- (topic, data) -> {
- if (data instanceof String) {
- return Serdes.String().serializer().serialize(topic, (String) data);
- }
- return Serdes.Double().serializer().serialize(topic, (Double) data);
- },
- processor);
-
- testDriver = new TopologyTestDriver(topology);
-
- final org.apache.kafka.streams.test.ConsumerRecordFactory<Long, String> source1Factory =
- new org.apache.kafka.streams.test.ConsumerRecordFactory<>(
- SOURCE_TOPIC_1,
- Serdes.Long().serializer(),
- Serdes.String().serializer());
- final org.apache.kafka.streams.test.ConsumerRecordFactory<Integer, Double> source2Factory =
- new org.apache.kafka.streams.test.ConsumerRecordFactory<>(
- SOURCE_TOPIC_2,
- Serdes.Integer().serializer(),
- Serdes.Double().serializer());
-
- final Long source1Key = 42L;
- final String source1Value = "anyString";
- final Integer source2Key = 73;
- final Double source2Value = 3.14;
-
- final ConsumerRecord<byte[], byte[]> consumerRecord1 = source1Factory.create(source1Key, source1Value);
- final ConsumerRecord<byte[], byte[]> consumerRecord2 = source2Factory.create(source2Key, source2Value);
-
- testDriver.pipeInput(consumerRecord1);
- final ProducerRecord<Long, String> record1 =
- testDriver.readOutput(SINK_TOPIC_1, Serdes.Long().deserializer(), Serdes.String().deserializer());
- assertThat(record1.key(), equalTo(source1Key));
- assertThat(record1.value(), equalTo(source1Value));
-
- testDriver.pipeInput(consumerRecord2);
- final ProducerRecord<Integer, Double> record2 =
- testDriver.readOutput(SINK_TOPIC_1, Serdes.Integer().deserializer(), Serdes.Double().deserializer());
- assertThat(record2.key(), equalTo(source2Key));
- assertThat(record2.value(), equalTo(source2Value));
- }
-
@Test
public void shouldUseSourceSpecificDeserializers() {
final Topology topology = new Topology();
@@ -848,33 +751,6 @@ public abstract class TopologyTestDriverTest {
assertThat(result2.getValue(), equalTo(source2Value));
}
- @Deprecated
- //Test not migrated to non-deprecated methods, List processing now in TestInputTopic
- @Test
- public void shouldProcessConsumerRecordList() {
- testDriver = new TopologyTestDriver(setupMultipleSourceTopology(SOURCE_TOPIC_1, SOURCE_TOPIC_2));
-
- final List<TTDTestRecord> processedRecords1 = mockProcessors.get(0).processedRecords;
- final List<TTDTestRecord> processedRecords2 = mockProcessors.get(1).processedRecords;
-
- final List<ConsumerRecord<byte[], byte[]>> testRecords = new ArrayList<>(2);
- testRecords.add(consumerRecord1);
- testRecords.add(consumerRecord2);
-
- testDriver.pipeInput(testRecords);
-
- assertEquals(1, processedRecords1.size());
- assertEquals(1, processedRecords2.size());
-
- TTDTestRecord record = processedRecords1.get(0);
- TTDTestRecord expectedResult = new TTDTestRecord(consumerRecord1, 0L);
- assertThat(record, equalTo(expectedResult));
-
- record = processedRecords2.get(0);
- expectedResult = new TTDTestRecord(consumerRecord2, 0L);
- assertThat(record, equalTo(expectedResult));
- }
-
@Test
public void shouldForwardRecordsFromSubtopologyToSubtopology() {
testDriver = new TopologyTestDriver(setupTopologyWithTwoSubtopologies());
@@ -954,36 +830,6 @@ public abstract class TopologyTestDriverTest {
assertThat(mockPunctuator.punctuatedAt, equalTo(expectedPunctuations));
}
- @SuppressWarnings("deprecation")
- //Testing already deprecated methods until methods removed
- @Test
- public void shouldPunctuateOnWallClockTimeDeprecated() {
- final MockPunctuator mockPunctuator = new MockPunctuator();
- testDriver = new TopologyTestDriver(
- setupSingleProcessorTopology(10L, PunctuationType.WALL_CLOCK_TIME, mockPunctuator),
- config,
- 0);
-
- final List<Long> expectedPunctuations = new LinkedList<>();
- testDriver.advanceWallClockTime(5L);
- assertThat(mockPunctuator.punctuatedAt, equalTo(expectedPunctuations));
-
- expectedPunctuations.add(14L);
- testDriver.advanceWallClockTime(9L);
- assertThat(mockPunctuator.punctuatedAt, equalTo(expectedPunctuations));
-
- testDriver.advanceWallClockTime(1L);
- assertThat(mockPunctuator.punctuatedAt, equalTo(expectedPunctuations));
-
- expectedPunctuations.add(35L);
- testDriver.advanceWallClockTime(20L);
- assertThat(mockPunctuator.punctuatedAt, equalTo(expectedPunctuations));
-
- expectedPunctuations.add(40L);
- testDriver.advanceWallClockTime(5L);
- assertThat(mockPunctuator.punctuatedAt, equalTo(expectedPunctuations));
- }
-
@Test
public void shouldPunctuateOnWallClockTime() {
final MockPunctuator mockPunctuator = new MockPunctuator();
@@ -1461,7 +1307,6 @@ public abstract class TopologyTestDriverTest {
ProcessorContext<String, Long> context;
private KeyValueStore<String, Long> store;
- @SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext<String, Long> context) {
this.context = context;
@@ -1517,7 +1362,6 @@ public abstract class TopologyTestDriverTest {
return new Processor<String, Long, Void, Void>() {
private KeyValueStore<String, Long> store;
- @SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext<Void, Void> context) {
this.store = context.getStateStore("storeProcessorStore");
@@ -1753,7 +1597,6 @@ public abstract class TopologyTestDriverTest {
() -> new Processor<String, String, Void, Void>() {
private KeyValueStore<String, String> stateStore;
- @SuppressWarnings("unchecked")
@Override
public void init(final ProcessorContext<Void, Void> context) {
stateStore = context.getStateStore("global-store");
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryTest.java
deleted file mode 100644
index deccdc9..0000000
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/test/ConsumerRecordFactoryTest.java
+++ /dev/null
@@ -1,273 +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.streams.test;
-
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.common.serialization.ByteArraySerializer;
-import org.apache.kafka.common.serialization.IntegerSerializer;
-import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.streams.KeyValue;
-import org.junit.jupiter.api.Test;
-
-import java.util.Arrays;
-import java.util.Collections;
-import java.util.List;
-
-import static org.junit.jupiter.api.Assertions.assertArrayEquals;
-import static org.junit.jupiter.api.Assertions.assertEquals;
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-@Deprecated
-public class ConsumerRecordFactoryTest {
- private final StringSerializer stringSerializer = new StringSerializer();
- private final IntegerSerializer integerSerializer = new IntegerSerializer();
-
- private final String topicName = "topic";
- private final String otherTopicName = "otherTopic";
- private final String key = "key";
- private final Integer value = 42;
- private final long timestamp = 21L;
- private final byte[] rawKey = stringSerializer.serialize(topicName, key);
- private final byte[] rawValue = integerSerializer.serialize(topicName, value);
-
- private final ConsumerRecordFactory<byte[], Integer> factory =
- new ConsumerRecordFactory<>(topicName, new ByteArraySerializer(), integerSerializer, 0L);
-
- private final ConsumerRecordFactory<byte[], Integer> defaultFactory =
- new ConsumerRecordFactory<>(new ByteArraySerializer(), integerSerializer);
-
- @Test
- public void shouldAdvanceTime() {
- factory.advanceTimeMs(3L);
- verifyRecord(topicName, rawKey, rawValue, 3L, factory.create(topicName, rawKey, value));
-
- factory.advanceTimeMs(2L);
- verifyRecord(topicName, rawKey, rawValue, 5L, factory.create(topicName, rawKey, value));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullTopicName() {
- assertThrows(NullPointerException.class, () -> factory.create(null, rawKey, value, timestamp));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullHeaders() {
- assertThrows(NullPointerException.class, () -> factory.create(topicName, rawKey, value, null, timestamp));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullTopicNameWithDefaultTimestamp() {
- assertThrows(NullPointerException.class, () -> factory.create(null, rawKey, value));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullTopicNameWithNullKey() {
- assertThrows(NullPointerException.class, () -> factory.create((String) null, value, timestamp));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullTopicNameWithNullKeyAndDefaultTimestamp() {
- assertThrows(NullPointerException.class, () -> factory.create((String) null, value));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullTopicNameWithKeyValuePairs() {
- assertThrows(NullPointerException.class, () -> factory.create(null, Collections.singletonList(KeyValue.pair(rawKey, value))));
- }
-
- @Test
- public void shouldNotAllowToCreateTopicWithNullTopicNameWithKeyValuePairsAndCustomTimestamps() {
- assertThrows(NullPointerException.class,
- () -> factory.create(null, Collections.singletonList(KeyValue.pair(rawKey, value)), timestamp, 2L));
- }
-
- @Test
- public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicName() {
- assertThrows(IllegalStateException.class, () -> defaultFactory.create(rawKey, value, timestamp));
- }
-
- @Test
- public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithDefaultTimestamp() {
- assertThrows(IllegalStateException.class, () -> defaultFactory.create(rawKey, value));
- }
-
- @Test
- public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithNullKey() {
- assertThrows(IllegalStateException.class, () -> defaultFactory.create(value, timestamp));
- }
-
- @Test
- public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithNullKeyAndDefaultTimestamp() {
- assertThrows(IllegalStateException.class, () -> defaultFactory.create(value));
- }
-
- @Test
- public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithKeyValuePairs() {
- assertThrows(IllegalStateException.class, () -> defaultFactory.create(Collections.singletonList(KeyValue.pair(rawKey, value))));
- }
-
- @Test
- public void shouldRequireCustomTopicNameIfNotDefaultFactoryTopicNameWithKeyValuePairsAndCustomTimestamps() {
- assertThrows(IllegalStateException.class, () -> defaultFactory.create(Collections.singletonList(KeyValue.pair(rawKey, value)), timestamp, 2L));
- }
-
- @Test
- public void shouldCreateConsumerRecordWithOtherTopicNameAndTimestamp() {
- verifyRecord(otherTopicName, rawKey, rawValue, timestamp, factory.create(otherTopicName, rawKey, value, timestamp));
- }
-
- @Test
- public void shouldCreateConsumerRecordWithTimestamp() {
- verifyRecord(topicName, rawKey, rawValue, timestamp, factory.create(rawKey, value, timestamp));
- }
-
- @Test
- public void shouldCreateConsumerRecordWithOtherTopicName() {
- verifyRecord(otherTopicName, rawKey, rawValue, 0L, factory.create(otherTopicName, rawKey, value));
-
- factory.advanceTimeMs(3L);
- verifyRecord(otherTopicName, rawKey, rawValue, 3L, factory.create(otherTopicName, rawKey, value));
- }
-
- @Test
- public void shouldCreateConsumerRecord() {
- verifyRecord(topicName, rawKey, rawValue, 0L, factory.create(rawKey, value));
-
- factory.advanceTimeMs(3L);
- verifyRecord(topicName, rawKey, rawValue, 3L, factory.create(rawKey, value));
- }
-
- @Test
- public void shouldCreateNullKeyConsumerRecordWithOtherTopicNameAndTimestampWithTimetamp() {
- verifyRecord(topicName, null, rawValue, timestamp, factory.create(value, timestamp));
- }
-
- @Test
- public void shouldCreateNullKeyConsumerRecordWithTimestampWithTimestamp() {
- verifyRecord(topicName, null, rawValue, timestamp, factory.create(value, timestamp));
- }
-
- @Test
- public void shouldCreateNullKeyConsumerRecord() {
- verifyRecord(topicName, null, rawValue, 0L, factory.create(value));
-
- factory.advanceTimeMs(3L);
- verifyRecord(topicName, null, rawValue, 3L, factory.create(value));
- }
-
- @SuppressWarnings("unchecked")
- @Test
- public void shouldCreateConsumerRecordsFromKeyValuePairs() {
- final ConsumerRecordFactory<String, Integer> factory =
- new ConsumerRecordFactory<>(topicName, stringSerializer, integerSerializer, 0L);
-
- final KeyValue[] keyValuePairs = new KeyValue[5];
- final KeyValue[] rawKeyValuePairs = new KeyValue[keyValuePairs.length];
-
- for (int i = 0; i < keyValuePairs.length; ++i) {
- keyValuePairs[i] = KeyValue.pair(key + "-" + i, value + i);
- rawKeyValuePairs[i] = KeyValue.pair(
- stringSerializer.serialize(topicName, key + "-" + i),
- integerSerializer.serialize(topicName, value + i));
- }
-
- final List<ConsumerRecord<byte[], byte[]>> records =
- factory.create(Arrays.<KeyValue<String, Integer>>asList(keyValuePairs));
-
- for (int i = 0; i < keyValuePairs.length; ++i) {
- verifyRecord(
- topicName,
- (byte[]) rawKeyValuePairs[i].key,
- (byte[]) rawKeyValuePairs[i].value,
- 0L,
- records.get(i));
- }
- }
-
- @SuppressWarnings("unchecked")
- @Test
- public void shouldCreateConsumerRecordsFromKeyValuePairsWithTimestampAndIncrements() {
- final ConsumerRecordFactory<String, Integer> factory =
- new ConsumerRecordFactory<>(topicName, stringSerializer, integerSerializer, timestamp, 2L);
-
- final KeyValue[] keyValuePairs = new KeyValue[5];
- final KeyValue[] rawKeyValuePairs = new KeyValue[keyValuePairs.length];
-
- for (int i = 0; i < keyValuePairs.length; ++i) {
- keyValuePairs[i] = KeyValue.pair(key + "-" + i, value + i);
- rawKeyValuePairs[i] = KeyValue.pair(
- stringSerializer.serialize(topicName, key + "-" + i),
- integerSerializer.serialize(topicName, value + i));
- }
-
- final List<ConsumerRecord<byte[], byte[]>> records =
- factory.create(Arrays.<KeyValue<String, Integer>>asList(keyValuePairs));
-
- for (int i = 0; i < keyValuePairs.length; ++i) {
- verifyRecord(
- topicName,
- (byte[]) rawKeyValuePairs[i].key,
- (byte[]) rawKeyValuePairs[i].value,
- timestamp + 2L * i,
- records.get(i));
- }
- }
-
- @SuppressWarnings("unchecked")
- @Test
- public void shouldCreateConsumerRecordsFromKeyValuePairsWithCustomTimestampAndIncrementsAndNotAdvanceTime() {
- final ConsumerRecordFactory<String, Integer> factory =
- new ConsumerRecordFactory<>(topicName, stringSerializer, integerSerializer, 0L);
-
- final KeyValue[] keyValuePairs = new KeyValue[5];
- final KeyValue[] rawKeyValuePairs = new KeyValue[keyValuePairs.length];
-
- for (int i = 0; i < keyValuePairs.length; ++i) {
- keyValuePairs[i] = KeyValue.pair(key + "-" + i, value + i);
- rawKeyValuePairs[i] = KeyValue.pair(
- stringSerializer.serialize(topicName, key + "-" + i),
- integerSerializer.serialize(topicName, value + i));
- }
-
- final List<ConsumerRecord<byte[], byte[]>> records =
- factory.create(Arrays.<KeyValue<String, Integer>>asList(keyValuePairs), timestamp, 2L);
-
- for (int i = 0; i < keyValuePairs.length; ++i) {
- verifyRecord(
- topicName,
- (byte[]) rawKeyValuePairs[i].key,
- (byte[]) rawKeyValuePairs[i].value,
- timestamp + 2L * i,
- records.get(i));
- }
-
- // should not have incremented internally tracked time
- verifyRecord(topicName, null, rawValue, 0L, factory.create(value));
- }
-
- private void verifyRecord(final String topicName,
- final byte[] rawKey,
- final byte[] rawValue,
- final long timestamp,
- final ConsumerRecord<byte[], byte[]> record) {
- assertEquals(topicName, record.topic());
- assertArrayEquals(rawKey, record.key());
- assertArrayEquals(rawValue, record.value());
- assertEquals(timestamp, record.timestamp());
- }
-
-}
diff --git a/streams/test-utils/src/test/java/org/apache/kafka/streams/test/OutputVerifierTest.java b/streams/test-utils/src/test/java/org/apache/kafka/streams/test/OutputVerifierTest.java
deleted file mode 100644
index 6aebe4f..0000000
--- a/streams/test-utils/src/test/java/org/apache/kafka/streams/test/OutputVerifierTest.java
+++ /dev/null
@@ -1,452 +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.streams.test;
-
-import org.apache.kafka.clients.producer.ProducerRecord;
-
-import org.junit.jupiter.api.Test;
-
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
-@Deprecated
-public class OutputVerifierTest {
- private final byte[] key = new byte[0];
- private final byte[] value = new byte[0];
-
- private final ProducerRecord<byte[], byte[]> producerRecord = new ProducerRecord<>(
- "someTopic",
- Integer.MAX_VALUE,
- Long.MAX_VALUE,
- key,
- value
- );
-
- private final ProducerRecord<byte[], byte[]> nullKeyValueRecord = new ProducerRecord<byte[], byte[]>(
- "someTopic",
- Integer.MAX_VALUE,
- Long.MAX_VALUE,
- null,
- null
- );
-
- @Test
- public void shouldNotAllowNullProducerRecordForCompareValue() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareValue(null, value));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordWithExpectedRecordForCompareValue() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareValue((ProducerRecord<byte[], byte[]>) null, producerRecord));
- }
-
- @Test
- public void shouldNotAllowNullExpectedRecordForCompareValue() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareValue(producerRecord, (ProducerRecord<byte[], byte[]>) null));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordForCompareKeyValue() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareKeyValue(null, key, value));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordWithExpectedRecordForCompareKeyValue() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareKeyValue(null, producerRecord));
- }
-
- @Test
- public void shouldNotAllowNullExpectedRecordForCompareKeyValue() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareKeyValue(producerRecord, null));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordForCompareValueTimestamp() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareValueTimestamp(null, value, 0L));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordWithExpectedRecordForCompareValueTimestamp() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareValueTimestamp(null, producerRecord));
- }
-
- @Test
- public void shouldNotAllowNullExpectedRecordForCompareValueTimestamp() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareValueTimestamp(producerRecord, null));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordForCompareKeyValueTimestamp() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareKeyValueTimestamp(null, key, value, 0L));
- }
-
- @Test
- public void shouldNotAllowNullProducerRecordWithExpectedRecordForCompareKeyValueTimestamp() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareKeyValueTimestamp(null, producerRecord));
- }
-
- @Test
- public void shouldNotAllowNullExpectedRecordForCompareKeyValueTimestamp() {
- assertThrows(NullPointerException.class, () -> OutputVerifier.compareKeyValueTimestamp(producerRecord, null));
- }
-
- @Test
- public void shouldPassIfValueIsEqualForCompareValue() {
- OutputVerifier.compareValue(producerRecord, value);
- }
-
- @Test
- public void shouldPassIfValueIsEqualWithNullForCompareValue() {
- OutputVerifier.compareValue(nullKeyValueRecord, (byte[]) null);
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValue(producerRecord, key));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValue(producerRecord, (byte[]) null));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReverseForCompareValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValue(nullKeyValueRecord, value));
- }
-
- @Test
- public void shouldPassIfValueIsEqualForCompareValueWithProducerRecord() {
- OutputVerifier.compareValue(producerRecord, new ProducerRecord<>(
- "otherTopic",
- 0,
- 0L,
- value,
- value
- ));
- }
-
- @Test
- public void shouldPassIfValueIsEqualWithNullForCompareValueWithProducerRecord() {
- OutputVerifier.compareValue(nullKeyValueRecord, new ProducerRecord<byte[], byte[]>(
- "otherTopic",
- 0,
- 0L,
- value,
- null
- ));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValue(producerRecord,
- new ProducerRecord<>("sameTopic", Integer.MAX_VALUE, Long.MAX_VALUE, value, key)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValue(producerRecord,
- new ProducerRecord<byte[], byte[]>("sameTopic", Integer.MAX_VALUE, Long.MAX_VALUE, value, null)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReverseForCompareValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValue(nullKeyValueRecord,
- new ProducerRecord<>("sameTopic", Integer.MAX_VALUE, Long.MAX_VALUE, value, value)));
- }
-
- @Test
- public void shouldPassIfKeyAndValueIsEqualForCompareKeyValue() {
- OutputVerifier.compareKeyValue(producerRecord, key, value);
- }
-
- @Test
- public void shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValue() {
- OutputVerifier.compareKeyValue(nullKeyValueRecord, null, null);
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentForCompareKeyValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord, value, value));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullForCompareKeyValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord, null, value));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, null, value), key, value));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareKeyValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord, key, key));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareKeyValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord, key, null));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReversForCompareKeyValue() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null), key, value));
- }
-
- @Test
- public void shouldPassIfKeyAndValueIsEqualForCompareKeyValueWithProducerRecord() {
- OutputVerifier.compareKeyValue(producerRecord, new ProducerRecord<>(
- "otherTopic",
- 0,
- 0L,
- key,
- value));
- }
-
- @Test
- public void shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueWithProducerRecord() {
- OutputVerifier.compareKeyValue(nullKeyValueRecord, new ProducerRecord<byte[], byte[]>(
- "otherTopic",
- 0,
- 0L,
- null,
- null));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentForCompareKeyValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, value, value)));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullForCompareKeyValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, null, value)));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, null, value), producerRecord));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareKeyValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, key)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareKeyValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(producerRecord,
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValue(
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null),
- producerRecord));
- }
-
- @Test
- public void shouldPassIfValueAndTimestampIsEqualForCompareValueTimestamp() {
- OutputVerifier.compareValueTimestamp(producerRecord, value, Long.MAX_VALUE);
- }
-
- @Test
- public void shouldPassIfValueAndTimestampIsEqualWithNullForCompareValueTimestamp() {
- OutputVerifier.compareValueTimestamp(nullKeyValueRecord, null, Long.MAX_VALUE);
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareValueTimestamp(producerRecord, key, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareValueTimestamp(producerRecord, null, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReverseForCompareValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareValueTimestamp(nullKeyValueRecord, value, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldPassIfValueAndTimestampIsEqualForCompareValueTimestampWithProducerRecord() {
- OutputVerifier.compareValueTimestamp(producerRecord, new ProducerRecord<>(
- "otherTopic",
- 0,
- Long.MAX_VALUE,
- value,
- value
- ));
- }
-
- @Test
- public void shouldPassIfValueAndTimestampIsEqualWithNullForCompareValueTimestampWithProducerRecord() {
- OutputVerifier.compareValueTimestamp(nullKeyValueRecord, new ProducerRecord<byte[], byte[]>(
- "otherTopic",
- 0,
- Long.MAX_VALUE,
- value,
- null
- ));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValueTimestamp(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, key)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValueTimestamp(producerRecord,
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReverseForCompareValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValueTimestamp(nullKeyValueRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, value)));
- }
-
- @Test
- public void shouldFailIfTimestampIsDifferentForCompareValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareValueTimestamp(producerRecord, value, 0));
- }
-
- @Test
- public void shouldFailIfTimestampDifferentWithNullReverseForCompareValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareValueTimestamp(nullKeyValueRecord, null, 0));
- }
-
- @Test
- public void shouldFailIfTimestampIsDifferentForCompareValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareValueTimestamp(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, 0L, key, value)));
- }
-
- @Test
- public void shouldPassIfKeyAndValueAndTimestampIsEqualForCompareKeyValueTimestamp() {
- OutputVerifier.compareKeyValueTimestamp(producerRecord, key, value, Long.MAX_VALUE);
- }
-
- @Test
- public void shouldPassIfKeyAndValueIsEqualWithNullForCompareKeyValueTimestamp() {
- OutputVerifier.compareKeyValueTimestamp(nullKeyValueRecord, null, null, Long.MAX_VALUE);
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentForCompareKeyValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareKeyValueTimestamp(producerRecord, value, value, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullForCompareKeyValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareKeyValueTimestamp(producerRecord, null, value, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueTimestamp() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, null, value),
- key, value, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareKeyValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareKeyValueTimestamp(producerRecord, key, key, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareKeyValueTimestamp() {
- assertThrows(AssertionError.class,
- () -> OutputVerifier.compareKeyValueTimestamp(producerRecord, key, null, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestamp() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null),
- key, value, Long.MAX_VALUE));
- }
-
- @Test
- public void shouldPassIfKeyAndValueAndTimestampIsEqualForCompareKeyValueTimestampWithProducerRecord() {
- OutputVerifier.compareKeyValueTimestamp(producerRecord, new ProducerRecord<>(
- "otherTopic", 0, Long.MAX_VALUE, key, value));
- }
-
- @Test
- public void shouldPassIfKeyAndValueAndTimestampIsEqualWithNullForCompareKeyValueTimestampWithProducerRecord() {
- OutputVerifier.compareKeyValueTimestamp(nullKeyValueRecord, new ProducerRecord<byte[], byte[]>(
- "otherTopic", 0, Long.MAX_VALUE, null, null));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentForCompareKeyValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, value, value)));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, null, value)));
- }
-
- @Test
- public void shouldFailIfKeyIsDifferentWithNullReversForCompareKeyValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, null, value), producerRecord));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentForCompareKeyValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(producerRecord,
- new ProducerRecord<>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, key)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullForCompareKeyValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(producerRecord,
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null)));
- }
-
- @Test
- public void shouldFailIfValueIsDifferentWithNullReversForCompareKeyValueTimestampWithProducerRecord() {
- assertThrows(AssertionError.class, () -> OutputVerifier.compareKeyValueTimestamp(
- new ProducerRecord<byte[], byte[]>("someTopic", Integer.MAX_VALUE, Long.MAX_VALUE, key, null), producerRecord));
- }
-
-}