From 3bb8c98102a8750663941508528ef96910a981c8 Mon Sep 17 00:00:00 2001 From: Steven van Rossum Date: Tue, 27 Oct 2026 10:52:57 +0100 Subject: [PATCH] Remove obsolete ConsumerSpEL and ProducerSpEL reflection shims in KafkaIO --- .../beam/sdk/io/kafka/ConsumerSpEL.java | 169 ------------------ .../sdk/io/kafka/KafkaExactlyOnceSink.java | 44 ++--- .../org/apache/beam/sdk/io/kafka/KafkaIO.java | 48 +---- .../beam/sdk/io/kafka/KafkaIOUtils.java | 25 +++ .../apache/beam/sdk/io/kafka/KafkaRecord.java | 5 - .../beam/sdk/io/kafka/KafkaRecordCoder.java | 32 ++-- .../sdk/io/kafka/KafkaUnboundedReader.java | 14 +- .../sdk/io/kafka/ProducerRecordCoder.java | 49 ++--- .../beam/sdk/io/kafka/ProducerSpEL.java | 157 ---------------- .../beam/sdk/io/kafka/ReadFromKafkaDoFn.java | 12 +- .../apache/beam/sdk/io/kafka/KafkaIOTest.java | 36 +--- .../sdk/io/kafka/KafkaRecordCoderTest.java | 4 +- .../sdk/io/kafka/ProducerRecordCoderTest.java | 24 +-- 13 files changed, 93 insertions(+), 526 deletions(-) delete mode 100644 sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ConsumerSpEL.java delete mode 100644 sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/ProducerSpEL.java 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 5045afb53be7..000000000000 --- 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 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 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 deserializeKey( - Deserializer deserializer, ConsumerRecord 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 deserializeValue( - Deserializer deserializer, ConsumerRecord 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 c6b7761a9a6c..6a811bc2e8fe 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.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 private final WriteRecords 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 spec) { this.spec = spec; } @@ -301,12 +295,6 @@ private static class ExactlyOnceWriter 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 @@ public void processElement( 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 @@ private static class ShardWriter { } void beginTxn() { - ProducerSpEL.beginTransaction(producer); + producer.beginTransaction(); } Future sendRecord( @@ -526,12 +514,15 @@ Future sendRecord( 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 @@ void commitTxn(long lastRecordId, Counter numTransactions) throws IOException { // 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 @@ void commitTxn(long lastRecordId, Counter numTransactions) throws IOException { 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 @@ private static Producer initializeExactlyOnceProducer( 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 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 a41db9fbeaf4..e5a7ce9e8830 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.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 PCollection> expand(PBegin input) { 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 PCollection> expand(PCollection 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 WriteRecords withPublishTimestampFunction( * written by the same job. */ public WriteRecords 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 PDone expand(PCollection> input) { 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 @@ private static ProducerRecord toProducerRecord( Long timestamp = row.getSchema().hasField(FIELD_TIMESTAMP) ? row.getInt64(FIELD_TIMESTAMP) : null; - boolean hasHeaders = ConsumerSpEL.hasHeaders(); Iterable
headers = Collections.emptyList(); - if (hasHeaders && row.getSchema().hasField(FIELD_HEADERS)) { + if (row.getSchema().hasField(FIELD_HEADERS)) { Iterable headerRows = row.getArray(FIELD_HEADERS); if (headerRows != null) { List
headerList = new ArrayList<>(); @@ -4146,16 +4109,9 @@ private static ProducerRecord toProducerRecord( } 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 91aa85577959..656eabfb52a1 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.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 @@ static Map overrideBootstrapServersConfig( 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 60aa68e0129c..e31333fc7e49 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 long getOffset() { @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 dbb3a053099c..cfccfe3674c2 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.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.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 extends StructuredCoder> { @@ -76,29 +74,21 @@ public KafkaRecord decode(InputStream inStream) throws IOException { 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> records) { - if (!ConsumerSpEL.hasHeaders()) { - return null; - } - - // ConsumerRecord is used to simply create a list of headers - ConsumerRecord consumerRecord = new ConsumerRecord<>("", 0, 0L, "", ""); - records.forEach(kv -> consumerRecord.headers().add(kv.getKey(), kv.getValue())); - return consumerRecord.headers(); + private static Headers toHeaders(Iterable> records) { + RecordHeaders headers = new RecordHeaders(); + records.forEach(kv -> headers.add(kv.getKey(), kv.getValue())); + return headers; } - private Iterable> toIterable(KafkaRecord record) { - if (!ConsumerSpEL.hasHeaders()) { - return Collections.emptyList(); - } - + private static Iterable> toIterable(KafkaRecord record) { List> 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 Object structuralValue(KafkaRecord value) { value.getOffset(), value.getTimestamp(), value.getTimestampType(), - !ConsumerSpEL.hasHeaders() ? null : value.getHeaders(), + value.getHeaders(), (KV) 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 5e18ad83e2ac..bd79fad56e6f 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 @@ public boolean advance() throws IOException { 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 @@ private void setupInitialOffset(PartitionState pState) { 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 f04868fa082c..63175546a32a 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.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.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 ProducerRecord decode(InputStream inStream) throws IOException { timestamp = null; } - Headers headers = (Headers) toHeaders(headerCoder.decode(inStream)); + Headers headers = toHeaders(headerCoder.decode(inStream)); KV kv = kvCoder.decode(inStream); - @SuppressWarnings("nullness") // kakfa library not annotated + @SuppressWarnings("nullness") // kafka library not annotated ProducerRecord 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> records) { - if (!ConsumerSpEL.hasHeaders()) { - return null; - } - - // ConsumerRecord is used to simply create a list of headers - ConsumerRecord consumerRecord = new ConsumerRecord<>("", 0, 0L, "", ""); - records.forEach(kv -> consumerRecord.headers().add(kv.getKey(), kv.getValue())); - return consumerRecord.headers(); + private static Headers toHeaders(Iterable> records) { + RecordHeaders headers = new RecordHeaders(); + records.forEach(kv -> headers.add(kv.getKey(), kv.getValue())); + return headers; } - private Iterable> toIterable(ProducerRecord record) { - if (!ConsumerSpEL.hasHeaders()) { - return Collections.emptyList(); - } + private static Iterable> toIterable(ProducerRecord record) { List> vals = new ArrayList<>(); for (Header header : record.headers()) { vals.add(KV.of(header.key(), header.value())); @@ -127,24 +116,18 @@ public boolean isRegisterByteSizeObserverCheap(ProducerRecord value) { return kvCoder.isRegisterByteSizeObserverCheap(KV.of(value.key(), value.value())); } - @SuppressWarnings("unchecked") @Override public Object structuralValue(ProducerRecord 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 c993f41421a1..000000000000 --- 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 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 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 684534434751..223548225c11 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 @@ public ProcessContinuation processElement( 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 8e56f08f7a1b..b3865162484e 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.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.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 @@ @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 static class IntegerDeserializerWithHeadersAssertor extends IntegerDeseri @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 static class LongDeserializerWithHeadersAssertor extends LongDeserializer @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 void testExactlyOnceSink() { // 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 void testExactlyOnceSink() { @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 void testSinkWithSendErrors() throws Throwable { @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 @@ numElements, new ValueAsTimestampFn(), false) @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 84d8cedb895a..50a5abddbdb5 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 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 void testKafkaRecordSerializableWithHeaders() throws IOException { @Test public void testKafkaRecordSerializableWithoutHeaders() throws IOException { - ConsumerRecord 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 6bc3fbaf02ac..47a3a5c3c4c3 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 @@ 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.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 void testProducerRecordSerializableWithHeaders() throws IOException { @Test public void testProducerRecordSerializableWithoutHeaders() throws IOException { - ConsumerRecord consumerRecord = new ConsumerRecord<>("", 0, 0L, "", ""); - verifySerialization(consumerRecord.headers(), 0, System.currentTimeMillis()); + verifySerialization(new RecordHeaders(), 0, System.currentTimeMillis()); } @Test @@ -102,24 +98,6 @@ public void testProducerRecordStructuralValueWithHeadersApi() throws IOException assertEquals(testProducerRecord.headers(), headers); } - @Test - public void testProducerRecordStructuralValueWithoutHeadersApi() throws IOException { - RecordHeaders headers = new RecordHeaders(); - headers.add("headerKey", "headerVal".getBytes(UTF_8)); - ProducerRecordCoder producerRecordCoder = - ProducerRecordCoder.of(ByteArrayCoder.of(), ByteArrayCoder.of()); - ProducerRecord producerRecord = - new ProducerRecord<>( - "topic", 1, null, "key".getBytes(UTF_8), "value".getBytes(UTF_8), headers); - try (MockedStatic staticMock = mockStatic(ConsumerSpEL.class)) { - staticMock.when(ConsumerSpEL::hasHeaders).thenReturn(false); - ProducerRecord testProducerRecord = - (ProducerRecord) producerRecordCoder.structuralValue(producerRecord); - - assertEquals(testProducerRecord.headers(), new RecordHeaders()); - } - } - private ProducerRecord verifySerialization(Integer partition, Long timestamp) throws IOException { return verifySerialization(null, partition, timestamp);