This is an automated email from the ASF dual-hosted git repository.
Abacn pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/beam.git
The following commit(s) were added to refs/heads/master by this push:
new 0b90e18f396 Remove obsolete ConsumerSpEL and ProducerSpEL reflection
shims in KafkaIO (#40839)
0b90e18f396 is described below
commit 0b90e18f396c157054c568b39eb4cce8ac9b0ac4
Author: Steven van Rossum <[email protected]>
AuthorDate: Tue Oct 27 17:40:34 2026 +0100
Remove obsolete ConsumerSpEL and ProducerSpEL reflection shims in KafkaIO
(#40839)
---
.../org/apache/beam/sdk/io/kafka/ConsumerSpEL.java | 169 ---------------------
.../beam/sdk/io/kafka/KafkaExactlyOnceSink.java | 44 +++---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 48 +-----
.../org/apache/beam/sdk/io/kafka/KafkaIOUtils.java | 25 +++
.../org/apache/beam/sdk/io/kafka/KafkaRecord.java | 5 -
.../apache/beam/sdk/io/kafka/KafkaRecordCoder.java | 32 ++--
.../beam/sdk/io/kafka/KafkaUnboundedReader.java | 14 +-
.../beam/sdk/io/kafka/ProducerRecordCoder.java | 49 ++----
.../org/apache/beam/sdk/io/kafka/ProducerSpEL.java | 157 -------------------
.../beam/sdk/io/kafka/ReadFromKafkaDoFn.java | 12 +-
.../org/apache/beam/sdk/io/kafka/KafkaIOTest.java | 36 +----
.../beam/sdk/io/kafka/KafkaRecordCoderTest.java | 4 +-
.../beam/sdk/io/kafka/ProducerRecordCoderTest.java | 24 +--
13 files changed, 93 insertions(+), 526 deletions(-)
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
deleted file mode 100644
index 5045afb53be..00000000000
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java
+++ /dev/null
@@ -1,169 +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.beam.sdk.io.kafka;
-
-import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
-
-import java.util.Map;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
-import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.header.Headers;
-import org.apache.kafka.common.serialization.Deserializer;
-import org.joda.time.Instant;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * ConsumerSpEL to handle multiple of versions of Consumer API between Kafka
0.9 and 2.1.0 onwards.
- * It auto detects the input type List/Collection/Varargs, to eliminate the
method definition
- * differences.
- */
-class ConsumerSpEL {
-
- private static final Logger LOG =
LoggerFactory.getLogger(ConsumerSpEL.class);
-
- private static boolean hasRecordTimestamp;
- private static boolean hasHeaders;
- private static boolean hasOffsetsForTimes;
- private static boolean deserializerSupportsHeaders;
-
- static {
- try {
- // It is supported by Kafka Client 0.10.0.0 onwards.
- hasRecordTimestamp =
- ConsumerRecord.class
- .getMethod("timestamp", (Class<?>[]) null)
- .getReturnType()
- .equals(Long.TYPE);
- } catch (NoSuchMethodException | SecurityException e) {
- LOG.debug("Timestamp for Kafka message is not available.");
- }
-
- try {
- // It is supported by Kafka Client 0.11.0.0 onwards.
- hasHeaders =
- "org.apache.kafka.common.header.Headers"
- .equals(
- ConsumerRecord.class
- .getMethod("headers", (Class<?>[]) null)
- .getReturnType()
- .getName());
- } catch (NoSuchMethodException | SecurityException e) {
- LOG.debug("Headers is not available");
- }
-
- try {
- // It is supported by Kafka Client 0.10.1.0 onwards.
- hasOffsetsForTimes =
- Consumer.class.getMethod("offsetsForTimes",
Map.class).getReturnType().equals(Map.class);
- } catch (NoSuchMethodException | SecurityException e) {
- LOG.debug("OffsetsForTimes is not available.");
- }
-
- try {
- // It is supported by Kafka Client 2.1.0 onwards.
- deserializerSupportsHeaders =
- "T"
- .equals(
- Deserializer.class
- .getDeclaredMethod("deserialize", String.class,
Headers.class, byte[].class)
- .getGenericReturnType()
- .getTypeName());
- } catch (NoSuchMethodException | SecurityException e) {
- LOG.debug("Deserializer interface does not support Kafka headers");
- }
- }
-
- public static long getRecordTimestamp(ConsumerRecord<byte[], byte[]>
rawRecord) {
- if (hasRecordTimestamp) {
- return rawRecord.timestamp();
- }
- return -1L; // This is the timestamp used in Kafka for older messages
without timestamps.
- }
-
- public static KafkaTimestampType getRecordTimestampType(
- ConsumerRecord<byte[], byte[]> rawRecord) {
- if (hasRecordTimestamp) {
- return
KafkaTimestampType.forOrdinal(rawRecord.timestampType().ordinal());
- } else {
- return KafkaTimestampType.NO_TIMESTAMP_TYPE;
- }
- }
-
- public static boolean hasOffsetsForTimes() {
- return hasOffsetsForTimes;
- }
-
- public static boolean hasHeaders() {
- return hasHeaders;
- }
-
- public static boolean deserializerSupportsHeaders() {
- return deserializerSupportsHeaders;
- }
-
- public static <T> T deserializeKey(
- Deserializer<T> deserializer, ConsumerRecord<byte[], byte[]> rawRecord) {
- if (deserializerSupportsHeaders) {
- // Kafka API 2.1.0 onwards
- return deserializer.deserialize(rawRecord.topic(), rawRecord.headers(),
rawRecord.key());
- } else {
- return deserializer.deserialize(rawRecord.topic(), rawRecord.key());
- }
- }
-
- public static <T> T deserializeValue(
- Deserializer<T> deserializer, ConsumerRecord<byte[], byte[]> rawRecord) {
- if (deserializerSupportsHeaders) {
- // Kafka API 2.1.0 onwards
- return deserializer.deserialize(rawRecord.topic(), rawRecord.headers(),
rawRecord.value());
- } else {
- return deserializer.deserialize(rawRecord.topic(), rawRecord.value());
- }
- }
-
- /**
- * Look up the offset for the given partition by timestamp. Throws
RuntimeException if there are
- * no messages later than timestamp or if this partition does not support
timestamp based offset.
- */
- @SuppressWarnings("unchecked")
- public static long offsetForTime(
- Consumer<?, ?> consumer, TopicPartition topicPartition, Instant time) {
-
- checkArgument(hasOffsetsForTimes, "This Kafka Client must support
Consumer.OffsetsForTimes().");
-
- // 'value' in the map returned by offsetFoTime() is null if there is no
offset for the time.
- OffsetAndTimestamp offsetAndTimestamp =
- Iterables.getOnlyElement(
- consumer.offsetsForTimes(ImmutableMap.of(topicPartition,
time.getMillis())).values());
-
- if (offsetAndTimestamp == null) {
- throw new RuntimeException(
- "There are no messages has a timestamp that is greater than or "
- + "equals to the target time or the message format version in
this partition is "
- + "before 0.10.0, topicPartition is: "
- + topicPartition);
- } else {
- return offsetAndTimestamp.offset();
- }
- }
-}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java
index c6b7761a9a6..6a811bc2e8f 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaExactlyOnceSink.java
@@ -17,7 +17,6 @@
*/
package org.apache.beam.sdk.io.kafka;
-import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkArgument;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkNotNull;
import static
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.base.Preconditions.checkState;
@@ -83,6 +82,9 @@ import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.clients.producer.RecordMetadata;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.TopicPartition;
+import org.apache.kafka.common.errors.AuthorizationException;
+import org.apache.kafka.common.errors.OutOfOrderSequenceException;
+import org.apache.kafka.common.errors.ProducerFencedException;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.joda.time.DateTimeUtils;
@@ -148,14 +150,6 @@ class KafkaExactlyOnceSink<K, V>
private final WriteRecords<K, V> spec;
- static void ensureEOSSupport() {
- checkArgument(
- ProducerSpEL.supportsTransactions(),
- "%s %s",
- "This version of Kafka client does not support transactions required
to support",
- "exactly-once semantics. Please use Kafka client version 0.11 or
newer.");
- }
-
KafkaExactlyOnceSink(WriteRecords<K, V> spec) {
this.spec = spec;
}
@@ -301,12 +295,6 @@ class KafkaExactlyOnceSink<K, V>
StateSpecs.bag(KvCoder.of(BigEndianLongCoder.of(),
TimestampedValueCoder.of(elemCoder)));
}
- @Setup
- public void setup() {
- // This is on the worker. Ensure the runtime version is till compatible.
- KafkaExactlyOnceSink.ensureEOSSupport();
- }
-
// Futures ignored as exceptions will be flushed out in the commitTxn
@SuppressWarnings("FutureReturnValueIgnored")
@RequiresStableInput
@@ -428,7 +416,7 @@ class KafkaExactlyOnceSink<K, V>
writer.commitTxn(nextId - 1, numTransactions);
nextIdState.write(nextId);
- } catch (ProducerSpEL.UnrecoverableProducerException e) {
+ } catch (ProducerFencedException | OutOfOrderSequenceException |
AuthorizationException e) {
// Producer JavaDoc says these are not recoverable errors and producer
should be closed.
// Close the producer and a new producer will be initialized in retry.
@@ -501,7 +489,7 @@ class KafkaExactlyOnceSink<K, V>
}
void beginTxn() {
- ProducerSpEL.beginTransaction(producer);
+ producer.beginTransaction();
}
Future<RecordMetadata> sendRecord(
@@ -526,12 +514,15 @@ class KafkaExactlyOnceSink<K, V>
record.getValue().value()));
sendCounter.inc();
return result;
+ } catch (ProducerFencedException | OutOfOrderSequenceException |
AuthorizationException e) {
+ throw e;
} catch (KafkaException e) {
- ProducerSpEL.abortTransaction(producer);
+ producer.abortTransaction();
throw e;
}
}
+ @SuppressWarnings("deprecation")
void commitTxn(long lastRecordId, Counter numTransactions) throws
IOException {
String topic = Preconditions.checkStateNotNull(spec.getTopic());
try {
@@ -539,15 +530,14 @@ class KafkaExactlyOnceSink<K, V>
// NOTE: Kafka keeps this metadata for 24 hours since the last
update. This limits
// how long the pipeline could be down before resuming it. It does
not look like
// this TTL can be adjusted (asked about it on Kafka users list).
- ProducerSpEL.sendOffsetsToTransaction(
- producer,
+ producer.sendOffsetsToTransaction(
ImmutableMap.of(
new TopicPartition(topic, shard),
new OffsetAndMetadata(
0L,
JSON_MAPPER.writeValueAsString(new
ShardMetadata(lastRecordId, writerId)))),
- spec.getSinkGroupId());
- ProducerSpEL.commitTransaction(producer);
+ Preconditions.checkStateNotNull(spec.getSinkGroupId()));
+ producer.commitTransaction();
numTransactions.inc();
if (!reportedLineage) {
@@ -567,8 +557,10 @@ class KafkaExactlyOnceSink<K, V>
LOG.debug("{} : committed {} records", shard, lastRecordId -
committedId);
committedId = lastRecordId;
+ } catch (ProducerFencedException | OutOfOrderSequenceException |
AuthorizationException e) {
+ throw e;
} catch (KafkaException e) {
- ProducerSpEL.abortTransaction(producer);
+ producer.abortTransaction();
throw e;
}
}
@@ -767,15 +759,15 @@ class KafkaExactlyOnceSink<K, V>
if (spec.getValueSerializer() != null) {
producerConfig.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG,
spec.getValueSerializer());
}
- producerConfig.put(ProducerSpEL.ENABLE_IDEMPOTENCE_CONFIG, true);
- producerConfig.put(ProducerSpEL.TRANSACTIONAL_ID_CONFIG, producerName);
+ producerConfig.put(ProducerConfig.ENABLE_IDEMPOTENCE_CONFIG, true);
+ producerConfig.put(ProducerConfig.TRANSACTIONAL_ID_CONFIG, producerName);
Producer<K, V> producer =
spec.getProducerFactoryFn() != null
? spec.getProducerFactoryFn().apply(producerConfig)
: new KafkaProducer<>(producerConfig);
- ProducerSpEL.initTransactions(producer);
+ producer.initTransactions();
return producer;
}
}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
index a41db9fbeaf..e5a7ce9e883 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIO.java
@@ -142,7 +142,6 @@ import
org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.apache.kafka.common.serialization.Deserializer;
import org.apache.kafka.common.serialization.Serializer;
import org.apache.kafka.common.serialization.StringSerializer;
-import org.apache.kafka.common.utils.AppInfoParser;
import org.checkerframework.checker.nullness.qual.MonotonicNonNull;
import org.checkerframework.checker.nullness.qual.Nullable;
import org.checkerframework.dataflow.qual.Pure;
@@ -1664,31 +1663,6 @@ public class KafkaIO {
checkArgument(getKeyDeserializerProvider() != null,
"withKeyDeserializer() is required");
checkArgument(getValueDeserializerProvider() != null,
"withValueDeserializer() is required");
- if (!ConsumerSpEL.hasOffsetsForTimes()) {
- LOG.warn(
- "Kafka client version {} is too old. Versions before 0.10.1.0 are
deprecated and "
- + "may not be supported in next release of Apache Beam. "
- + "Please upgrade your Kafka client version.",
- AppInfoParser.getVersion());
- }
- if (getStartReadTime() != null) {
- checkArgument(
- ConsumerSpEL.hasOffsetsForTimes(),
- "Consumer.offsetsForTimes is only supported by Kafka Client
0.10.1.0 onwards, "
- + "current version of Kafka Client is "
- + AppInfoParser.getVersion()
- + ". If you are building with maven, set
\"kafka.clients.version\" "
- + "maven property to 0.10.1.0 or newer.");
- }
- if (getStopReadTime() != null) {
- checkArgument(
- ConsumerSpEL.hasOffsetsForTimes(),
- "Consumer.offsetsForTimes is only supported by Kafka Client
0.10.1.0 onwards, "
- + "current version of Kafka Client is "
- + AppInfoParser.getVersion()
- + ". If you are building with maven, set
\"kafka.clients.version\" "
- + "maven property to 0.10.1.0 or newer.");
- }
if (isCommitOffsetsInFinalizeEnabled()) {
checkArgument(
getConsumerConfig().get(ConsumerConfig.GROUP_ID_CONFIG) != null,
@@ -3123,14 +3097,6 @@ public class KafkaIO {
checkArgument(getKeyDeserializerProvider() != null,
"withKeyDeserializer() is required");
checkArgument(getValueDeserializerProvider() != null,
"withValueDeserializer() is required");
- if (!ConsumerSpEL.hasOffsetsForTimes()) {
- LOG.warn(
- "Kafka client version {} is too old. Versions before 0.10.1.0 are
deprecated and "
- + "may not be supported in next release of Apache Beam. "
- + "Please upgrade your Kafka client version.",
- AppInfoParser.getVersion());
- }
-
if (isCommitOffsetEnabled()) {
if (configuredKafkaCommit()) {
LOG.info(
@@ -3529,7 +3495,6 @@ public class KafkaIO {
* written by the same job.
*/
public WriteRecords<K, V> withEOS(int numShards, String sinkGroupId) {
- KafkaExactlyOnceSink.ensureEOSSupport();
checkArgument(numShards >= 1, "numShards should be >= 1");
checkArgument(sinkGroupId != null, "sinkGroupId is required for
exactly-once sink");
return
toBuilder().setEOS(true).setNumShards(numShards).setSinkGroupId(sinkGroupId).build();
@@ -3581,7 +3546,6 @@ public class KafkaIO {
checkArgument(
getBadRecordErrorHandler() instanceof DefaultErrorHandler,
"BadRecordErrorHandling isn't supported with Kafka Exactly Once
writing");
- KafkaExactlyOnceSink.ensureEOSSupport();
// TODO: Verify that the group_id does not have existing state stored
on Kafka unless
// this is an upgrade. This avoids issues with simple mistake of
reusing group_id
@@ -4132,9 +4096,8 @@ public class KafkaIO {
Long timestamp =
row.getSchema().hasField(FIELD_TIMESTAMP) ?
row.getInt64(FIELD_TIMESTAMP) : null;
- boolean hasHeaders = ConsumerSpEL.hasHeaders();
Iterable<Header> headers = Collections.emptyList();
- if (hasHeaders && row.getSchema().hasField(FIELD_HEADERS)) {
+ if (row.getSchema().hasField(FIELD_HEADERS)) {
Iterable<Row> headerRows = row.getArray(FIELD_HEADERS);
if (headerRows != null) {
List<Header> headerList = new ArrayList<>();
@@ -4146,16 +4109,9 @@ public class KafkaIO {
}
headers = headerList;
}
- } else if (!hasHeaders && row.getSchema().hasField(FIELD_HEADERS)) {
- // Log warning when headers are present but Kafka client doesn't
support them
- LOG.warn(
- "Dropping headers from Kafka record because the Kafka client
version "
- + "does not support headers (requires Kafka 0.11+).");
}
- return hasHeaders
- ? new ProducerRecord<>(topic, partition, timestamp, key, value,
headers)
- : new ProducerRecord<>(topic, partition, timestamp, key, value);
+ return new ProducerRecord<>(topic, partition, timestamp, key, value,
headers);
}
}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOUtils.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOUtils.java
index 91aa8557795..656eabfb52a 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOUtils.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOUtils.java
@@ -27,12 +27,16 @@ import java.util.Random;
import java.util.concurrent.atomic.AtomicLongFieldUpdater;
import org.apache.beam.sdk.transforms.SerializableFunction;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.ImmutableMap;
+import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.collect.Iterables;
import
org.apache.beam.vendor.guava.v32_1_2_jre.com.google.common.primitives.Longs;
import org.apache.kafka.clients.consumer.Consumer;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.KafkaConsumer;
+import org.apache.kafka.clients.consumer.OffsetAndTimestamp;
+import org.apache.kafka.common.TopicPartition;
import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.checkerframework.checker.nullness.qual.Nullable;
+import org.joda.time.Instant;
/**
* Common utility functions and default configurations for {@link
KafkaIO.Read} and {@link
@@ -148,6 +152,27 @@ public final class KafkaIOUtils {
return config;
}
+ /**
+ * Look up the offset for the given partition by timestamp. Throws
RuntimeException if there are
+ * no messages later than timestamp or if this partition does not support
timestamp based offset.
+ */
+ static long offsetForTime(Consumer<?, ?> consumer, TopicPartition
topicPartition, Instant time) {
+ // 'value' in the map returned by offsetsForTimes() is null if there is no
offset for the time.
+ OffsetAndTimestamp offsetAndTimestamp =
+ Iterables.getOnlyElement(
+ consumer.offsetsForTimes(ImmutableMap.of(topicPartition,
time.getMillis())).values());
+
+ if (offsetAndTimestamp == null) {
+ throw new RuntimeException(
+ "There are no messages has a timestamp that is greater than or "
+ + "equals to the target time or the message format version in
this partition is "
+ + "before 0.10.0, topicPartition is: "
+ + topicPartition);
+ } else {
+ return offsetAndTimestamp.offset();
+ }
+ }
+
/*
* Maintains approximate average over last 1000 elements.
* Usage is only thread-safe for a single producer and multiple consumers.
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecord.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecord.java
index 60aa68e0129..e31333fc7e4 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecord.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecord.java
@@ -84,11 +84,6 @@ public class KafkaRecord<K, V> {
@Pure
public @Nullable Headers getHeaders() {
- if (!ConsumerSpEL.hasHeaders()) {
- throw new RuntimeException(
- "The version kafka-clients does not support record headers, "
- + "please use version 0.11.0.0 or newer");
- }
return headers;
}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoder.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoder.java
index dbb3a053099..cfccfe3674c 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoder.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoder.java
@@ -21,7 +21,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.Coder;
@@ -33,10 +32,9 @@ import org.apache.beam.sdk.coders.StructuredCoder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.values.KV;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Header;
import org.apache.kafka.common.header.Headers;
-import org.checkerframework.checker.nullness.qual.Nullable;
+import org.apache.kafka.common.header.internals.RecordHeaders;
/** {@link Coder} for {@link KafkaRecord}. */
public class KafkaRecordCoder<K, V> extends StructuredCoder<KafkaRecord<K, V>>
{
@@ -76,29 +74,21 @@ public class KafkaRecordCoder<K, V> extends
StructuredCoder<KafkaRecord<K, V>> {
longCoder.decode(inStream),
longCoder.decode(inStream),
KafkaTimestampType.forOrdinal(intCoder.decode(inStream)),
- (Headers) toHeaders(headerCoder.decode(inStream)),
+ toHeaders(headerCoder.decode(inStream)),
kvCoder.decode(inStream));
}
- private @Nullable Object toHeaders(Iterable<KV<String, byte[]>> records) {
- if (!ConsumerSpEL.hasHeaders()) {
- return null;
- }
-
- // ConsumerRecord is used to simply create a list of headers
- ConsumerRecord<String, String> consumerRecord = new ConsumerRecord<>("",
0, 0L, "", "");
- records.forEach(kv -> consumerRecord.headers().add(kv.getKey(),
kv.getValue()));
- return consumerRecord.headers();
+ private static Headers toHeaders(Iterable<KV<String, byte[]>> records) {
+ RecordHeaders headers = new RecordHeaders();
+ records.forEach(kv -> headers.add(kv.getKey(), kv.getValue()));
+ return headers;
}
- private Iterable<KV<String, byte[]>> toIterable(KafkaRecord<K, V> record) {
- if (!ConsumerSpEL.hasHeaders()) {
- return Collections.emptyList();
- }
-
+ private static Iterable<KV<String, byte[]>> toIterable(KafkaRecord<?, ?>
record) {
List<KV<String, byte[]>> vals = new ArrayList<>();
- if (record.getHeaders() != null) {
- for (Header header : record.getHeaders()) {
+ Headers headers = record.getHeaders();
+ if (headers != null) {
+ for (Header header : headers) {
vals.add(KV.of(header.key(), header.value()));
}
}
@@ -133,7 +123,7 @@ public class KafkaRecordCoder<K, V> extends
StructuredCoder<KafkaRecord<K, V>> {
value.getOffset(),
value.getTimestamp(),
value.getTimestampType(),
- !ConsumerSpEL.hasHeaders() ? null : value.getHeaders(),
+ value.getHeaders(),
(KV<Object, Object>) kvCoder.structuralValue(value.getKV()));
}
}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
index 5e18ad83e2a..bd79fad56e6 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaUnboundedReader.java
@@ -187,11 +187,13 @@ class KafkaUnboundedReader<K, V> extends
UnboundedReader<KafkaRecord<K, V>> {
rawRecord.topic(),
rawRecord.partition(),
rawRecord.offset(),
- ConsumerSpEL.getRecordTimestamp(rawRecord),
- ConsumerSpEL.getRecordTimestampType(rawRecord),
- ConsumerSpEL.hasHeaders() ? rawRecord.headers() : null,
- ConsumerSpEL.deserializeKey(keyDeserializerInstance,
rawRecord),
- ConsumerSpEL.deserializeValue(valueDeserializerInstance,
rawRecord));
+ rawRecord.timestamp(),
+
KafkaTimestampType.forOrdinal(rawRecord.timestampType().ordinal()),
+ rawRecord.headers(),
+ keyDeserializerInstance.deserialize(
+ rawRecord.topic(), rawRecord.headers(), rawRecord.key()),
+ valueDeserializerInstance.deserialize(
+ rawRecord.topic(), rawRecord.headers(),
rawRecord.value()));
curTimestamp =
pState.timestampPolicy.getTimestampForRecord(pState.mkTimestampPolicyContext(),
record);
@@ -828,7 +830,7 @@ class KafkaUnboundedReader<K, V> extends
UnboundedReader<KafkaRecord<K, V>> {
Instant startReadTime = spec.getStartReadTime();
if (startReadTime != null) {
pState.nextOffset =
- ConsumerSpEL.offsetForTime(consumer, pState.topicPartition,
startReadTime);
+ KafkaIOUtils.offsetForTime(consumer, pState.topicPartition,
startReadTime);
consumer.seek(pState.topicPartition, pState.nextOffset);
} else {
pState.nextOffset = consumer.position(pState.topicPartition);
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoder.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoder.java
index f04868fa082..63175546a32 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoder.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoder.java
@@ -21,7 +21,6 @@ import java.io.IOException;
import java.io.InputStream;
import java.io.OutputStream;
import java.util.ArrayList;
-import java.util.Collections;
import java.util.List;
import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.Coder;
@@ -32,10 +31,10 @@ import org.apache.beam.sdk.coders.StructuredCoder;
import org.apache.beam.sdk.coders.VarIntCoder;
import org.apache.beam.sdk.coders.VarLongCoder;
import org.apache.beam.sdk.values.KV;
-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;
+import org.apache.kafka.common.header.internals.RecordHeaders;
import org.checkerframework.checker.nullness.qual.Nullable;
/** {@link Coder} for {@link ProducerRecord}. */
@@ -78,33 +77,23 @@ public class ProducerRecordCoder<K, V> extends
StructuredCoder<ProducerRecord<K,
timestamp = null;
}
- Headers headers = (Headers) toHeaders(headerCoder.decode(inStream));
+ Headers headers = toHeaders(headerCoder.decode(inStream));
KV<K, V> kv = kvCoder.decode(inStream);
- @SuppressWarnings("nullness") // kakfa library not annotated
+ @SuppressWarnings("nullness") // kafka library not annotated
ProducerRecord<K, V> result =
- ConsumerSpEL.hasHeaders()
- ? new ProducerRecord<>(topic, partition, timestamp, kv.getKey(),
kv.getValue(), headers)
- : new ProducerRecord<>(topic, partition, timestamp, kv.getKey(),
kv.getValue());
+ new ProducerRecord<>(topic, partition, timestamp, kv.getKey(),
kv.getValue(), headers);
return result;
}
- private @Nullable Object toHeaders(Iterable<KV<String, byte[]>> records) {
- if (!ConsumerSpEL.hasHeaders()) {
- return null;
- }
-
- // ConsumerRecord is used to simply create a list of headers
- ConsumerRecord<String, String> consumerRecord = new ConsumerRecord<>("",
0, 0L, "", "");
- records.forEach(kv -> consumerRecord.headers().add(kv.getKey(),
kv.getValue()));
- return consumerRecord.headers();
+ private static Headers toHeaders(Iterable<KV<String, byte[]>> records) {
+ RecordHeaders headers = new RecordHeaders();
+ records.forEach(kv -> headers.add(kv.getKey(), kv.getValue()));
+ return headers;
}
- private Iterable<KV<String, byte[]>> toIterable(ProducerRecord<K, V> record)
{
- if (!ConsumerSpEL.hasHeaders()) {
- return Collections.emptyList();
- }
+ private static Iterable<KV<String, byte[]>> toIterable(ProducerRecord<?, ?>
record) {
List<KV<String, byte[]>> vals = new ArrayList<>();
for (Header header : record.headers()) {
vals.add(KV.of(header.key(), header.value()));
@@ -127,24 +116,18 @@ public class ProducerRecordCoder<K, V> extends
StructuredCoder<ProducerRecord<K,
return kvCoder.isRegisterByteSizeObserverCheap(KV.of(value.key(),
value.value()));
}
- @SuppressWarnings("unchecked")
@Override
public Object structuralValue(ProducerRecord<K, V> value) {
if (consistentWithEquals()) {
return value;
} else {
- if (!ConsumerSpEL.hasHeaders()) {
- return new ProducerRecord<>(
- value.topic(), value.partition(), value.timestamp(), value.key(),
value.value());
- } else {
- return new ProducerRecord<>(
- value.topic(),
- value.partition(),
- value.timestamp(),
- value.key(),
- value.value(),
- value.headers());
- }
+ return new ProducerRecord<>(
+ value.topic(),
+ value.partition(),
+ value.timestamp(),
+ value.key(),
+ value.value(),
+ value.headers());
}
}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerSpEL.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerSpEL.java
deleted file mode 100644
index c993f41421a..00000000000
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerSpEL.java
+++ /dev/null
@@ -1,157 +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.beam.sdk.io.kafka;
-
-import java.lang.reflect.InvocationTargetException;
-import java.lang.reflect.Method;
-import java.util.Map;
-import org.apache.beam.sdk.util.Preconditions;
-import org.apache.kafka.clients.consumer.OffsetAndMetadata;
-import org.apache.kafka.clients.producer.Producer;
-import org.apache.kafka.common.TopicPartition;
-import org.apache.kafka.common.errors.ApiException;
-import org.apache.kafka.common.errors.AuthorizationException;
-import org.checkerframework.checker.nullness.qual.Nullable;
-
-/**
- * ProducerSpEL to handle newer versions Producer API. The API is updated in
Kafka 0.11 to support
- * exactly-once semantics.
- */
-class ProducerSpEL {
-
- private static final class TransactionsImplementation {
- private Method initTransactionsMethod;
- private Method beginTransactionMethod;
- private Method commitTransactionMethod;
- private Method abortTransactionMethod;
- private Method sendOffsetsToTransactionMethod;
- private Class<?> producerFencedExceptionClass;
- private Class<?> outOfOrderSequenceExceptionClass;
-
- private TransactionsImplementation() throws NoSuchMethodException,
ClassNotFoundException {
- initTransactionsMethod = Producer.class.getMethod("initTransactions");
- beginTransactionMethod = Producer.class.getMethod("beginTransaction");
- commitTransactionMethod = Producer.class.getMethod("commitTransaction");
- abortTransactionMethod = Producer.class.getMethod("abortTransaction");
- sendOffsetsToTransactionMethod =
- Producer.class.getMethod("sendOffsetsToTransaction", Map.class,
String.class);
- producerFencedExceptionClass =
-
Class.forName("org.apache.kafka.common.errors.ProducerFencedException");
- outOfOrderSequenceExceptionClass =
-
Class.forName("org.apache.kafka.common.errors.OutOfOrderSequenceException");
- }
-
- private void initTransactions(Producer<?, ?> producer) {
- invoke(initTransactionsMethod, producer);
- }
-
- private void beginTransaction(Producer<?, ?> producer) {
- invoke(beginTransactionMethod, producer);
- }
-
- private void commitTransaction(Producer<?, ?> producer) {
- invoke(commitTransactionMethod, producer);
- }
-
- private void abortTransaction(Producer<?, ?> producer) {
- invoke(abortTransactionMethod, producer);
- }
-
- private void sendOffsetsToTransaction(
- Producer<?, ?> producer,
- Map<TopicPartition, OffsetAndMetadata> offsets,
- @Nullable String consumerGroupId) {
- invoke(sendOffsetsToTransactionMethod, producer, offsets,
consumerGroupId);
- }
-
- private void invoke(Method method, Object obj, @Nullable Object... args) {
- try {
- @SuppressWarnings({"nullness", "unused"}) // JDK annotation does not
allow the nulls
- Object ignored = method.invoke(obj, args);
- } catch (IllegalAccessException | InvocationTargetException e) {
- throw new RuntimeException(e);
- } catch (ApiException e) {
- Class<?> eClass = e.getClass();
- if (producerFencedExceptionClass.isAssignableFrom(eClass)
- || outOfOrderSequenceExceptionClass.isAssignableFrom(eClass)
- || AuthorizationException.class.isAssignableFrom(eClass)) {
- throw new UnrecoverableProducerException(e);
- }
- throw e;
- }
- }
- }
-
- private static @Nullable TransactionsImplementation
transactionsImplementation;
-
- static final String ENABLE_IDEMPOTENCE_CONFIG = "enable.idempotence";
- static final String TRANSACTIONAL_ID_CONFIG = "transactional.id";
-
- static {
- try {
- transactionsImplementation = new TransactionsImplementation();
- } catch (ClassNotFoundException | NoSuchMethodException e) {
- transactionsImplementation = null;
- }
- }
-
- /**
- * Wraps an unrecoverable producer exceptions, including the ones related
transactions introduced
- * in 0.11 (as described in documentation for {@link Producer}). The calller
should close the
- * producer when this exception is thrown.
- */
- static class UnrecoverableProducerException extends ApiException {
- UnrecoverableProducerException(ApiException cause) {
- super(cause);
- }
- }
-
- static boolean supportsTransactions() {
- return transactionsImplementation != null;
- }
-
- private static TransactionsImplementation getTransactionsImplementation() {
- return Preconditions.checkStateNotNull(
- transactionsImplementation,
- "This version of Kafka client library does not support transactions."
- + " Please used version 0.11 or later.");
- }
-
- static void initTransactions(Producer<?, ?> producer) {
- getTransactionsImplementation().initTransactions(producer);
- }
-
- static void beginTransaction(Producer<?, ?> producer) {
- getTransactionsImplementation().beginTransaction(producer);
- }
-
- static void commitTransaction(Producer<?, ?> producer) {
- getTransactionsImplementation().commitTransaction(producer);
- }
-
- static void abortTransaction(Producer<?, ?> producer) {
- getTransactionsImplementation().abortTransaction(producer);
- }
-
- static void sendOffsetsToTransaction(
- Producer<?, ?> producer,
- Map<TopicPartition, OffsetAndMetadata> offsets,
- @Nullable String consumerGroupId) {
- getTransactionsImplementation().sendOffsetsToTransaction(producer,
offsets, consumerGroupId);
- }
-}
diff --git
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
index 68453443475..223548225c1 100644
---
a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
+++
b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ReadFromKafkaDoFn.java
@@ -578,11 +578,13 @@ abstract class ReadFromKafkaDoFn<K, V>
rawRecord.topic(),
rawRecord.partition(),
rawRecord.offset(),
- ConsumerSpEL.getRecordTimestamp(rawRecord),
- ConsumerSpEL.getRecordTimestampType(rawRecord),
- ConsumerSpEL.hasHeaders() ? rawRecord.headers() : null,
- ConsumerSpEL.deserializeKey(keyDeserializerInstance,
rawRecord),
- ConsumerSpEL.deserializeValue(valueDeserializerInstance,
rawRecord));
+ rawRecord.timestamp(),
+
KafkaTimestampType.forOrdinal(rawRecord.timestampType().ordinal()),
+ rawRecord.headers(),
+ keyDeserializerInstance.deserialize(
+ rawRecord.topic(), rawRecord.headers(),
rawRecord.key()),
+ valueDeserializerInstance.deserialize(
+ rawRecord.topic(), rawRecord.headers(),
rawRecord.value()));
int recordSize =
(rawRecord.key() == null ? 0 : rawRecord.key().length)
+ (rawRecord.value() == null ? 0 :
rawRecord.value().length);
diff --git
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
index 8e56f08f7a1..b3865162484 100644
---
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
+++
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOTest.java
@@ -34,7 +34,6 @@ import static org.junit.Assert.assertFalse;
import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertNull;
import static org.junit.Assert.assertTrue;
-import static org.junit.Assume.assumeTrue;
import static org.junit.internal.matchers.ThrowableCauseMatcher.hasCause;
import static org.junit.internal.matchers.ThrowableMessageMatcher.hasMessage;
@@ -178,8 +177,6 @@ import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
import org.mockito.ArgumentCaptor;
import org.mockito.Mockito;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
/**
* Tests of {@link KafkaIO}. Run with 'mvn test
-Dkafka.clients.version=0.10.1.1', to test with a
@@ -189,8 +186,6 @@ import org.slf4j.LoggerFactory;
@RunWith(JUnit4.class)
public class KafkaIOTest {
- private static final Logger LOG = LoggerFactory.getLogger(KafkaIOTest.class);
-
/*
* The tests below borrow code and structure from CountingSourceTest. In
addition verifies
* the reader interleaves the records from multiple partitions.
@@ -645,14 +640,12 @@ public class KafkaIOTest {
@Override
public Integer deserialize(String topic, byte[] data) {
- assertEquals(false, ConsumerSpEL.deserializerSupportsHeaders());
- return super.deserialize(topic, data);
+ throw new UnsupportedOperationException(
+ "Expected deserialize(String, Headers, byte[]) to be called");
}
@Override
public Integer deserialize(String topic, Headers headers, byte[] data) {
- // Overriding the default should trigger header evaluation and this to
be called.
- assertEquals(true, ConsumerSpEL.deserializerSupportsHeaders());
return super.deserialize(topic, data);
}
}
@@ -662,14 +655,12 @@ public class KafkaIOTest {
@Override
public Long deserialize(String topic, byte[] data) {
- assertEquals(false, ConsumerSpEL.deserializerSupportsHeaders());
- return super.deserialize(topic, data);
+ throw new UnsupportedOperationException(
+ "Expected deserialize(String, Headers, byte[]) to be called");
}
@Override
public Long deserialize(String topic, Headers headers, byte[] data) {
- // Overriding the default should trigger header evaluation and this to
be called.
- assertEquals(true, ConsumerSpEL.deserializerSupportsHeaders());
return super.deserialize(topic, data);
}
}
@@ -2082,12 +2073,6 @@ public class KafkaIOTest {
// It is not feasible to test such retries with direct runner. When
DoFnTester supports
// state, we can test ExactlyOnceWriter DoFn directly to ensure it handles
retries correctly.
- if (!ProducerSpEL.supportsTransactions()) {
- LOG.warn(
- "testExactlyOnceSink() is disabled as Kafka client version does not
support transactions.");
- return;
- }
-
int numElements = 1000;
try (MockProducerWrapper producerWrapper = new MockProducerWrapper(new
LongSerializer())) {
@@ -2125,13 +2110,6 @@ public class KafkaIOTest {
@Test
public void testExactlyOnceSinkWithSendException() throws Throwable {
-
- if (!ProducerSpEL.supportsTransactions()) {
- LOG.warn(
- "testExactlyOnceSink() is disabled as Kafka client version does not
support transactions.");
- return;
- }
-
thrown.expect(KafkaException.class);
thrown.expectMessage("fakeException");
@@ -2201,9 +2179,6 @@ public class KafkaIOTest {
@Test
public void testUnboundedSourceStartReadTime() {
-
- assumeTrue(ConsumerSpEL.hasOffsetsForTimes());
-
int numElements = 1000;
// In this MockConsumer, we let the elements of the time and offset equal
and there are 20
// partitions. So set this startTime can read half elements.
@@ -2278,9 +2253,6 @@ public class KafkaIOTest {
@Test
public void testUnboundedSourceStartReadTimeException() {
-
- assumeTrue(ConsumerSpEL.hasOffsetsForTimes());
-
noMessagesException.expect(RuntimeException.class);
int numElements = 1000;
diff --git
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoderTest.java
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoderTest.java
index 84d8cedb895..50a5abddbdb 100644
---
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoderTest.java
+++
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaRecordCoderTest.java
@@ -26,7 +26,6 @@ import java.nio.charset.StandardCharsets;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.testing.CoderProperties;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.Test;
@@ -51,8 +50,7 @@ public class KafkaRecordCoderTest {
@Test
public void testKafkaRecordSerializableWithoutHeaders() throws IOException {
- ConsumerRecord<String, String> consumerRecord = new ConsumerRecord<>("",
0, 0L, "", "");
- verifySerialization(consumerRecord.headers());
+ verifySerialization(new RecordHeaders());
}
@Test
diff --git
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoderTest.java
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoderTest.java
index 6bc3fbaf02a..47a3a5c3c4c 100644
---
a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoderTest.java
+++
b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/ProducerRecordCoderTest.java
@@ -20,7 +20,6 @@ package org.apache.beam.sdk.io.kafka;
import static java.nio.charset.StandardCharsets.UTF_8;
import static org.junit.Assert.assertEquals;
import static org.junit.Assert.assertNull;
-import static org.mockito.Mockito.mockStatic;
import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
@@ -29,14 +28,12 @@ import org.apache.beam.sdk.coders.ByteArrayCoder;
import org.apache.beam.sdk.coders.StringUtf8Coder;
import org.apache.beam.sdk.testing.CoderProperties;
import org.apache.beam.sdk.transforms.windowing.GlobalWindow;
-import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.header.Headers;
import org.apache.kafka.common.header.internals.RecordHeaders;
import org.junit.Test;
import org.junit.runner.RunWith;
import org.junit.runners.JUnit4;
-import org.mockito.MockedStatic;
/** Tests for {@link ProducerRecordCoder}. */
@RunWith(JUnit4.class)
@@ -56,8 +53,7 @@ public class ProducerRecordCoderTest {
@Test
public void testProducerRecordSerializableWithoutHeaders() throws
IOException {
- ConsumerRecord<String, String> consumerRecord = new ConsumerRecord<>("",
0, 0L, "", "");
- verifySerialization(consumerRecord.headers(), 0,
System.currentTimeMillis());
+ verifySerialization(new RecordHeaders(), 0, System.currentTimeMillis());
}
@Test
@@ -102,24 +98,6 @@ public class ProducerRecordCoderTest {
assertEquals(testProducerRecord.headers(), headers);
}
- @Test
- public void testProducerRecordStructuralValueWithoutHeadersApi() throws
IOException {
- RecordHeaders headers = new RecordHeaders();
- headers.add("headerKey", "headerVal".getBytes(UTF_8));
- ProducerRecordCoder<byte[], byte[]> producerRecordCoder =
- ProducerRecordCoder.of(ByteArrayCoder.of(), ByteArrayCoder.of());
- ProducerRecord<byte[], byte[]> producerRecord =
- new ProducerRecord<>(
- "topic", 1, null, "key".getBytes(UTF_8), "value".getBytes(UTF_8),
headers);
- try (MockedStatic<ConsumerSpEL> staticMock =
mockStatic(ConsumerSpEL.class)) {
- staticMock.when(ConsumerSpEL::hasHeaders).thenReturn(false);
- ProducerRecord<byte[], byte[]> testProducerRecord =
- (ProducerRecord<byte[], byte[]>)
producerRecordCoder.structuralValue(producerRecord);
-
- assertEquals(testProducerRecord.headers(), new RecordHeaders());
- }
- }
-
private ProducerRecord<String, String> verifySerialization(Integer
partition, Long timestamp)
throws IOException {
return verifySerialization(null, partition, timestamp);
