You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@beam.apache.org by da...@apache.org on 2024/08/15 17:38:05 UTC
(beam) 01/01: Revert "Add in redistribute option for Kafka Read"
This is an automated email from the ASF dual-hosted git repository.
damccorm pushed a commit to branch revert-31347-kafka_alo
in repository https://gitbox.apache.org/repos/asf/beam.git
commit 7ade15924c3eaf48b5450a1af742df5f2174a81e
Author: Danny McCormick <da...@google.com>
AuthorDate: Thu Aug 15 19:37:57 2024 +0200
Revert "Add in redistribute option for Kafka Read"
---
.../java/org/apache/beam/sdk/io/kafka/KafkaIO.java | 158 +--------------------
.../KafkaIOReadImplementationCompatibility.java | 18 ---
.../beam/sdk/io/kafka/KafkaIOExternalTest.java | 17 +--
...KafkaIOReadImplementationCompatibilityTest.java | 4 +-
.../org/apache/beam/sdk/io/kafka/KafkaIOTest.java | 58 +-------
.../sdk/io/kafka/upgrade/KafkaIOTranslation.java | 22 ---
.../io/kafka/upgrade/KafkaIOTranslationTest.java | 1 -
sdks/python/apache_beam/io/kafka.py | 38 ++---
.../runners/portability/flink_runner_test.py | 3 -
9 files changed, 19 insertions(+), 300 deletions(-)
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 c1526d5382b..5e9ff9ab80c 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
@@ -75,7 +75,6 @@ import org.apache.beam.sdk.transforms.Impulse;
import org.apache.beam.sdk.transforms.MapElements;
import org.apache.beam.sdk.transforms.PTransform;
import org.apache.beam.sdk.transforms.ParDo;
-import org.apache.beam.sdk.transforms.Redistribute;
import org.apache.beam.sdk.transforms.Reshuffle;
import org.apache.beam.sdk.transforms.SerializableFunction;
import org.apache.beam.sdk.transforms.SimpleFunction;
@@ -601,9 +600,6 @@ public class KafkaIO {
.setDynamicRead(false)
.setTimestampPolicyFactory(TimestampPolicyFactory.withProcessingTime())
.setConsumerPollingTimeout(2L)
- .setRedistributed(false)
- .setAllowDuplicates(false)
- .setRedistributeNumKeys(0)
.build();
}
@@ -702,15 +698,6 @@ public class KafkaIO {
@Pure
public abstract boolean isDynamicRead();
- @Pure
- public abstract boolean isRedistributed();
-
- @Pure
- public abstract boolean isAllowDuplicates();
-
- @Pure
- public abstract int getRedistributeNumKeys();
-
@Pure
public abstract @Nullable Duration getWatchTopicPartitionDuration();
@@ -770,12 +757,6 @@ public class KafkaIO {
abstract Builder<K, V> setWatchTopicPartitionDuration(Duration duration);
- abstract Builder<K, V> setRedistributed(boolean withRedistribute);
-
- abstract Builder<K, V> setAllowDuplicates(boolean allowDuplicates);
-
- abstract Builder<K, V> setRedistributeNumKeys(int redistributeNumKeys);
-
abstract Builder<K, V> setTimestampPolicyFactory(
TimestampPolicyFactory<K, V> timestampPolicyFactory);
@@ -871,22 +852,6 @@ public class KafkaIO {
} else {
builder.setConsumerPollingTimeout(2L);
}
-
- if (config.redistribute != null) {
- builder.setRedistributed(config.redistribute);
- if (config.redistributeNumKeys != null) {
- builder.setRedistributeNumKeys((int) config.redistributeNumKeys);
- }
- if (config.allowDuplicates != null) {
- builder.setAllowDuplicates(config.allowDuplicates);
- }
-
- } else {
- builder.setRedistributed(false);
- builder.setRedistributeNumKeys(0);
- builder.setAllowDuplicates(false);
- }
- System.out.println("xxx builder service" + builder.toString());
}
private static <T> Coder<T> resolveCoder(Class<Deserializer<T>> deserializer) {
@@ -951,9 +916,6 @@ public class KafkaIO {
private Boolean commitOffsetInFinalize;
private Long consumerPollingTimeout;
private String timestampPolicy;
- private Integer redistributeNumKeys;
- private Boolean redistribute;
- private Boolean allowDuplicates;
public void setConsumerConfig(Map<String, String> consumerConfig) {
this.consumerConfig = consumerConfig;
@@ -998,18 +960,6 @@ public class KafkaIO {
public void setConsumerPollingTimeout(Long consumerPollingTimeout) {
this.consumerPollingTimeout = consumerPollingTimeout;
}
-
- public void setRedistributeNumKeys(Integer redistributeNumKeys) {
- this.redistributeNumKeys = redistributeNumKeys;
- }
-
- public void setRedistribute(Boolean redistribute) {
- this.redistribute = redistribute;
- }
-
- public void setAllowDuplicates(Boolean allowDuplicates) {
- this.allowDuplicates = allowDuplicates;
- }
}
}
@@ -1057,30 +1007,6 @@ public class KafkaIO {
return toBuilder().setTopicPartitions(ImmutableList.copyOf(topicPartitions)).build();
}
- /**
- * Sets redistribute transform that hints to the runner to try to redistribute the work evenly.
- */
- public Read<K, V> withRedistribute() {
- if (getRedistributeNumKeys() == 0 && isRedistributed()) {
- LOG.warn("This will create a key per record, which is sub-optimal for most use cases.");
- }
- return toBuilder().setRedistributed(true).build();
- }
-
- public Read<K, V> withAllowDuplicates(Boolean allowDuplicates) {
- if (!isAllowDuplicates()) {
- LOG.warn("Setting this value without setting withRedistribute() will have no effect.");
- }
- return toBuilder().setAllowDuplicates(allowDuplicates).build();
- }
-
- public Read<K, V> withRedistributeNumKeys(int redistributeNumKeys) {
- checkState(
- isRedistributed(),
- "withRedistributeNumKeys is ignored if withRedistribute() is not enabled on the transform.");
- return toBuilder().setRedistributeNumKeys(redistributeNumKeys).build();
- }
-
/**
* Internally sets a {@link java.util.regex.Pattern} of topics to read from. All the partitions
* from each of the matching topics are read.
@@ -1692,25 +1618,6 @@ public class KafkaIO {
.withMaxNumRecords(kafkaRead.getMaxNumRecords());
}
- if (kafkaRead.isRedistributed()) {
- // fail here instead.
- checkArgument(
- kafkaRead.isCommitOffsetsInFinalizeEnabled(),
- "commitOffsetsInFinalize() can't be enabled with isRedistributed");
- PCollection<KafkaRecord<K, V>> output = input.getPipeline().apply(transform);
- if (kafkaRead.getRedistributeNumKeys() == 0) {
- return output.apply(
- "Insert Redistribute",
- Redistribute.<KafkaRecord<K, V>>arbitrarily()
- .withAllowDuplicates(kafkaRead.isAllowDuplicates()));
- } else {
- return output.apply(
- "Insert Redistribute with Shards",
- Redistribute.<KafkaRecord<K, V>>arbitrarily()
- .withAllowDuplicates(kafkaRead.isAllowDuplicates())
- .withNumBuckets((int) kafkaRead.getRedistributeNumKeys()));
- }
- }
return input.getPipeline().apply(transform);
}
}
@@ -1743,15 +1650,6 @@ public class KafkaIO {
readTransform =
readTransform.withBadRecordErrorHandler(kafkaRead.getBadRecordErrorHandler());
}
- if (kafkaRead.isRedistributed()) {
- readTransform = readTransform.withRedistribute();
- }
- if (kafkaRead.isAllowDuplicates()) {
- readTransform = readTransform.withAllowDuplicates();
- }
- if (kafkaRead.getRedistributeNumKeys() > 0) {
- readTransform = readTransform.withRedistributeNumKeys(kafkaRead.getRedistributeNumKeys());
- }
PCollection<KafkaSourceDescriptor> output;
if (kafkaRead.isDynamicRead()) {
Set<String> topics = new HashSet<>();
@@ -1781,22 +1679,6 @@ public class KafkaIO {
.apply(Impulse.create())
.apply(ParDo.of(new GenerateKafkaSourceDescriptor(kafkaRead)));
}
- if (kafkaRead.isRedistributed()) {
- PCollection<KafkaRecord<K, V>> pcol =
- output.apply(readTransform).setCoder(KafkaRecordCoder.of(keyCoder, valueCoder));
- if (kafkaRead.getRedistributeNumKeys() == 0) {
- return pcol.apply(
- "Insert Redistribute",
- Redistribute.<KafkaRecord<K, V>>arbitrarily()
- .withAllowDuplicates(kafkaRead.isAllowDuplicates()));
- } else {
- return pcol.apply(
- "Insert Redistribute with Shards",
- Redistribute.<KafkaRecord<K, V>>arbitrarily()
- .withAllowDuplicates(true)
- .withNumBuckets((int) kafkaRead.getRedistributeNumKeys()));
- }
- }
return output.apply(readTransform).setCoder(KafkaRecordCoder.of(keyCoder, valueCoder));
}
}
@@ -2188,15 +2070,6 @@ public class KafkaIO {
@Pure
abstract boolean isCommitOffsetEnabled();
- @Pure
- abstract boolean isRedistribute();
-
- @Pure
- abstract boolean isAllowDuplicates();
-
- @Pure
- abstract int getRedistributeNumKeys();
-
@Pure
abstract @Nullable TimestampPolicyFactory<K, V> getTimestampPolicyFactory();
@@ -2263,12 +2136,6 @@ public class KafkaIO {
abstract ReadSourceDescriptors.Builder<K, V> setBounded(boolean bounded);
- abstract ReadSourceDescriptors.Builder<K, V> setRedistribute(boolean withRedistribute);
-
- abstract ReadSourceDescriptors.Builder<K, V> setAllowDuplicates(boolean allowDuplicates);
-
- abstract ReadSourceDescriptors.Builder<K, V> setRedistributeNumKeys(int redistributeNumKeys);
-
abstract ReadSourceDescriptors<K, V> build();
}
@@ -2281,9 +2148,6 @@ public class KafkaIO {
.setBadRecordRouter(BadRecordRouter.THROWING_ROUTER)
.setBadRecordErrorHandler(new ErrorHandler.DefaultErrorHandler<>())
.setConsumerPollingTimeout(2L)
- .setRedistribute(false)
- .setAllowDuplicates(false)
- .setRedistributeNumKeys(0)
.build()
.withProcessingTime()
.withMonotonicallyIncreasingWatermarkEstimator();
@@ -2443,19 +2307,6 @@ public class KafkaIO {
ReadSourceDescriptors.ExtractOutputTimestampFns.useProcessingTime());
}
- /** Enable Redistribute. */
- public ReadSourceDescriptors<K, V> withRedistribute() {
- return toBuilder().setRedistribute(true).build();
- }
-
- public ReadSourceDescriptors<K, V> withAllowDuplicates() {
- return toBuilder().setAllowDuplicates(true).build();
- }
-
- public ReadSourceDescriptors<K, V> withRedistributeNumKeys(int redistributeNumKeys) {
- return toBuilder().setRedistributeNumKeys(redistributeNumKeys).build();
- }
-
/** Use the creation time of {@link KafkaRecord} as the output timestamp. */
public ReadSourceDescriptors<K, V> withCreateTime() {
return withExtractOutputTimestampFn(
@@ -2646,12 +2497,6 @@ public class KafkaIO {
}
}
- if (isRedistribute()) {
- if (getRedistributeNumKeys() == 0) {
- LOG.warn("This will create a key per record, which is sub-optimal for most use cases.");
- }
- }
-
if (getConsumerConfig().get(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG) == null) {
LOG.warn(
"The bootstrapServers is not set. It must be populated through the KafkaSourceDescriptor during runtime otherwise the pipeline will fail.");
@@ -2682,7 +2527,7 @@ public class KafkaIO {
.getSchemaRegistry()
.getSchemaCoder(KafkaSourceDescriptor.class),
recordCoder));
- if (isCommitOffsetEnabled() && !configuredKafkaCommit() && !isRedistribute()) {
+ if (isCommitOffsetEnabled() && !configuredKafkaCommit()) {
outputWithDescriptor =
outputWithDescriptor
.apply(Reshuffle.viaRandomKey())
@@ -2693,7 +2538,6 @@ public class KafkaIO {
.getSchemaRegistry()
.getSchemaCoder(KafkaSourceDescriptor.class),
recordCoder));
-
PCollection<Void> unused = outputWithDescriptor.apply(new KafkaCommitOffset<K, V>(this));
unused.setCoder(VoidCoder.of());
}
diff --git a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibility.java b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibility.java
index 457e0003705..ce0434ee88d 100644
--- a/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibility.java
+++ b/sdks/java/io/kafka/src/main/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibility.java
@@ -119,24 +119,6 @@ class KafkaIOReadImplementationCompatibility {
return Long.valueOf(2);
}
},
- REDISTRIBUTE_NUM_KEYS {
- @Override
- Object getDefaultValue() {
- return Integer.valueOf(0);
- }
- },
- REDISTRIBUTED {
- @Override
- Object getDefaultValue() {
- return false;
- }
- },
- ALLOW_DUPLICATES {
- @Override
- Object getDefaultValue() {
- return false;
- }
- },
;
private final @NonNull ImmutableSet<KafkaIOReadImplementation> supportedImplementations;
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOExternalTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOExternalTest.java
index f021789a912..246fdd80d73 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOExternalTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOExternalTest.java
@@ -108,10 +108,7 @@ public class KafkaIOExternalTest {
Field.of("start_read_time", FieldType.INT64),
Field.of("commit_offset_in_finalize", FieldType.BOOLEAN),
Field.of("timestamp_policy", FieldType.STRING),
- Field.of("consumer_polling_timeout", FieldType.INT64),
- Field.of("redistribute_num_keys", FieldType.INT32),
- Field.of("redistribute", FieldType.BOOLEAN),
- Field.of("allow_duplicates", FieldType.BOOLEAN)))
+ Field.of("consumer_polling_timeout", FieldType.INT64)))
.withFieldValue("topics", topics)
.withFieldValue("consumer_config", consumerConfig)
.withFieldValue("key_deserializer", keyDeserializer)
@@ -120,9 +117,6 @@ public class KafkaIOExternalTest {
.withFieldValue("commit_offset_in_finalize", false)
.withFieldValue("timestamp_policy", "ProcessingTime")
.withFieldValue("consumer_polling_timeout", 5L)
- .withFieldValue("redistribute_num_keys", 0)
- .withFieldValue("redistribute", false)
- .withFieldValue("allow_duplicates", false)
.build());
RunnerApi.Components defaultInstance = RunnerApi.Components.getDefaultInstance();
@@ -145,7 +139,6 @@ public class KafkaIOExternalTest {
expansionService.expand(request, observer);
ExpansionApi.ExpansionResponse result = observer.result;
RunnerApi.PTransform transform = result.getTransform();
- System.out.println("xxx : " + result.toString());
assertThat(
transform.getSubtransformsList(),
Matchers.hasItem(MatchesPattern.matchesPattern(".*KafkaIO-Read.*")));
@@ -244,10 +237,7 @@ public class KafkaIOExternalTest {
Field.of("value_deserializer", FieldType.STRING),
Field.of("start_read_time", FieldType.INT64),
Field.of("commit_offset_in_finalize", FieldType.BOOLEAN),
- Field.of("timestamp_policy", FieldType.STRING),
- Field.of("redistribute_num_keys", FieldType.INT32),
- Field.of("redistribute", FieldType.BOOLEAN),
- Field.of("allow_duplicates", FieldType.BOOLEAN)))
+ Field.of("timestamp_policy", FieldType.STRING)))
.withFieldValue("topics", topics)
.withFieldValue("consumer_config", consumerConfig)
.withFieldValue("key_deserializer", keyDeserializer)
@@ -255,9 +245,6 @@ public class KafkaIOExternalTest {
.withFieldValue("start_read_time", startReadTime)
.withFieldValue("commit_offset_in_finalize", false)
.withFieldValue("timestamp_policy", "ProcessingTime")
- .withFieldValue("redistribute_num_keys", 0)
- .withFieldValue("redistribute", false)
- .withFieldValue("allow_duplicates", false)
.build());
RunnerApi.Components defaultInstance = RunnerApi.Components.getDefaultInstance();
diff --git a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
index ae939d66c21..2d306b0d779 100644
--- a/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
+++ b/sdks/java/io/kafka/src/test/java/org/apache/beam/sdk/io/kafka/KafkaIOReadImplementationCompatibilityTest.java
@@ -103,9 +103,7 @@ public class KafkaIOReadImplementationCompatibilityTest {
private void testReadTransformCreationWithImplementationBoundProperties(
Function<KafkaIO.Read<Integer, Long>, KafkaIO.Read<Integer, Long>> kafkaReadDecorator) {
- p.apply(
- kafkaReadDecorator.apply(
- mkKafkaReadTransform(1000, null, new ValueAsTimestampFn(), false, 0)));
+ p.apply(kafkaReadDecorator.apply(mkKafkaReadTransform(1000, null, new ValueAsTimestampFn())));
p.run();
}
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 73aee5aeeef..07e5b519c01 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
@@ -376,7 +376,7 @@ public class KafkaIOTest {
static KafkaIO.Read<Integer, Long> mkKafkaReadTransform(
int numElements, @Nullable SerializableFunction<KV<Integer, Long>, Instant> timestampFn) {
- return mkKafkaReadTransform(numElements, numElements, timestampFn, false, 0);
+ return mkKafkaReadTransform(numElements, numElements, timestampFn);
}
/**
@@ -386,9 +386,7 @@ public class KafkaIOTest {
static KafkaIO.Read<Integer, Long> mkKafkaReadTransform(
int numElements,
@Nullable Integer maxNumRecords,
- @Nullable SerializableFunction<KV<Integer, Long>, Instant> timestampFn,
- @Nullable Boolean redistribute,
- @Nullable Integer numKeys) {
+ @Nullable SerializableFunction<KV<Integer, Long>, Instant> timestampFn) {
List<String> topics = ImmutableList.of("topic_a", "topic_b");
@@ -406,16 +404,10 @@ public class KafkaIOTest {
}
if (timestampFn != null) {
- reader = reader.withTimestampFn(timestampFn);
+ return reader.withTimestampFn(timestampFn);
+ } else {
+ return reader;
}
-
- if (redistribute) {
- if (numKeys != null) {
- reader = reader.withRedistribute().withRedistributeNumKeys(numKeys);
- }
- reader = reader.withRedistribute();
- }
- return reader;
}
private static class AssertMultipleOf implements SerializableFunction<Iterable<Long>, Void> {
@@ -624,42 +616,6 @@ public class KafkaIOTest {
p.run();
}
- @Test
- public void testCommitOffsetsInFinalizeAndRedistributeErrors() {
- thrown.expect(Exception.class);
- thrown.expectMessage("commitOffsetsInFinalize() can't be enabled with isRedistributed");
-
- int numElements = 1000;
-
- PCollection<Long> input =
- p.apply(
- mkKafkaReadTransform(numElements, numElements, new ValueAsTimestampFn(), true, 0)
- .withConsumerConfigUpdates(
- ImmutableMap.of(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true))
- .withoutMetadata())
- .apply(Values.create());
-
- addCountingAsserts(input, numElements);
- p.run();
- }
-
- @Test
- public void testNumKeysIgnoredWithRedistributeNotEnabled() {
- int numElements = 1000;
-
- PCollection<Long> input =
- p.apply(
- mkKafkaReadTransform(numElements, numElements, new ValueAsTimestampFn(), false, 0)
- .withConsumerConfigUpdates(
- ImmutableMap.of(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG, true))
- .withoutMetadata())
- .apply(Values.create());
-
- addCountingAsserts(input, numElements);
-
- p.run();
- }
-
@Test
public void testUnreachableKafkaBrokers() {
// Expect an exception when the Kafka brokers are not reachable on the workers.
@@ -1949,7 +1905,7 @@ public class KafkaIOTest {
PCollection<Long> input =
p.apply(
- mkKafkaReadTransform(numElements, maxNumRecords, new ValueAsTimestampFn(), false, 0)
+ mkKafkaReadTransform(numElements, maxNumRecords, new ValueAsTimestampFn())
.withStartReadTime(new Instant(startTime))
.withoutMetadata())
.apply(Values.create());
@@ -1973,7 +1929,7 @@ public class KafkaIOTest {
int startTime = numElements / 20;
p.apply(
- mkKafkaReadTransform(numElements, numElements, new ValueAsTimestampFn(), false, 0)
+ mkKafkaReadTransform(numElements, numElements, new ValueAsTimestampFn())
.withStartReadTime(new Instant(startTime))
.withoutMetadata())
.apply(Values.create());
diff --git a/sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java b/sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java
index db7b172170a..9848e429e21 100644
--- a/sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java
+++ b/sdks/java/io/kafka/upgrade/src/main/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslation.java
@@ -98,9 +98,6 @@ public class KafkaIOTranslation {
.addNullableLogicalTypeField("stop_read_time", new NanosInstant())
.addBooleanField("is_commit_offset_finalize_enabled")
.addBooleanField("is_dynamic_read")
- .addBooleanField("redistribute")
- .addBooleanField("allows_duplicates")
- .addNullableInt32Field("redistribute_num_keys")
.addNullableLogicalTypeField("watch_topic_partition_duration", new NanosDuration())
.addByteArrayField("timestamp_policy_factory")
.addNullableMapField("offset_consumer_config", FieldType.STRING, FieldType.BYTES)
@@ -218,9 +215,6 @@ public class KafkaIOTranslation {
+ " is not supported yet.");
}
- fieldValues.put("redistribute", transform.isRedistributed());
- fieldValues.put("redistribute_num_keys", transform.getRedistributeNumKeys());
- fieldValues.put("allows_duplicates", transform.isAllowDuplicates());
return Row.withSchema(schema).withFieldValues(fieldValues).build();
}
@@ -331,22 +325,6 @@ public class KafkaIOTranslation {
if (maxNumRecords != null) {
transform = transform.withMaxNumRecords(maxNumRecords);
}
-
- Boolean isRedistributed = configRow.getBoolean("redistribute");
- if (isRedistributed != null && isRedistributed) {
- transform = transform.withRedistribute();
- Integer redistributeNumKeys =
- configRow.getValue("redistribute_num_keys") == null
- ? Integer.valueOf(0)
- : configRow.getInt32("redistribute_num_keys");
- if (redistributeNumKeys != null && !redistributeNumKeys.equals(0)) {
- transform = transform.withRedistributeNumKeys(redistributeNumKeys);
- }
- Boolean allowDuplicates = configRow.getBoolean("allows_duplicates");
- if (allowDuplicates != null && allowDuplicates) {
- transform = transform.withAllowDuplicates(allowDuplicates);
- }
- }
Duration maxReadTime = configRow.getValue("max_read_time");
if (maxReadTime != null) {
transform =
diff --git a/sdks/java/io/kafka/upgrade/src/test/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslationTest.java b/sdks/java/io/kafka/upgrade/src/test/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslationTest.java
index 095702a5c6f..f69b9c3649b 100644
--- a/sdks/java/io/kafka/upgrade/src/test/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslationTest.java
+++ b/sdks/java/io/kafka/upgrade/src/test/java/org/apache/beam/sdk/io/kafka/upgrade/KafkaIOTranslationTest.java
@@ -64,7 +64,6 @@ public class KafkaIOTranslationTest {
READ_TRANSFORM_SCHEMA_MAPPING.put("getMaxReadTime", "max_read_time");
READ_TRANSFORM_SCHEMA_MAPPING.put("getStartReadTime", "start_read_time");
READ_TRANSFORM_SCHEMA_MAPPING.put("getStopReadTime", "stop_read_time");
- READ_TRANSFORM_SCHEMA_MAPPING.put("getRedistributeNumKeys", "redistribute_num_keys");
READ_TRANSFORM_SCHEMA_MAPPING.put(
"isCommitOffsetsInFinalizeEnabled", "is_commit_offset_finalize_enabled");
READ_TRANSFORM_SCHEMA_MAPPING.put("isDynamicRead", "is_dynamic_read");
diff --git a/sdks/python/apache_beam/io/kafka.py b/sdks/python/apache_beam/io/kafka.py
index b19e9c22aa3..b4fd7d86e68 100644
--- a/sdks/python/apache_beam/io/kafka.py
+++ b/sdks/python/apache_beam/io/kafka.py
@@ -82,29 +82,19 @@
import typing
-import numpy as np
-
from apache_beam.transforms.external import BeamJarExpansionService
from apache_beam.transforms.external import ExternalTransform
from apache_beam.transforms.external import NamedTupleBasedPayloadBuilder
ReadFromKafkaSchema = typing.NamedTuple(
'ReadFromKafkaSchema',
- [
- ('consumer_config', typing.Mapping[str, str]),
- ('topics', typing.List[str]),
- ('key_deserializer', str),
- ('value_deserializer', str),
- ('start_read_time', typing.Optional[int]),
- ('max_num_records', typing.Optional[int]),
- ('max_read_time', typing.Optional[int]),
- ('commit_offset_in_finalize', bool),
- ('timestamp_policy', str),
- ('consumer_polling_timeout', typing.Optional[int]),
- ('redistribute', typing.Optional[bool]),
- ('redistribute_num_keys', typing.Optional[np.int32]),
- ('allow_duplicates', typing.Optional[bool]),
- ])
+ [('consumer_config', typing.Mapping[str, str]),
+ ('topics', typing.List[str]), ('key_deserializer', str),
+ ('value_deserializer', str), ('start_read_time', typing.Optional[int]),
+ ('max_num_records', typing.Optional[int]),
+ ('max_read_time', typing.Optional[int]),
+ ('commit_offset_in_finalize', bool), ('timestamp_policy', str),
+ ('consumer_polling_timeout', typing.Optional[int])])
def default_io_expansion_service(append_args=None):
@@ -148,9 +138,6 @@ class ReadFromKafka(ExternalTransform):
consumer_polling_timeout=2,
with_metadata=False,
expansion_service=None,
- redistribute=False,
- redistribute_num_keys=np.int32(0),
- allow_duplicates=False,
):
"""
Initializes a read operation from Kafka.
@@ -185,12 +172,6 @@ class ReadFromKafka(ExternalTransform):
this only works when using default key and value deserializers where
Java Kafka Reader reads keys and values as 'byte[]'.
:param expansion_service: The address (host:port) of the ExpansionService.
- :param redistribute: whether a Redistribute transform should be applied
- immediately after the read.
- :param redistribute_num_keys: Configures how many keys the Redistribute
- spreads the data across.
- :param allow_duplicates: whether the Redistribute transform allows for
- duplicates (this serves solely as a hint to the underlying runner).
"""
if timestamp_policy not in [ReadFromKafka.processing_time_policy,
ReadFromKafka.create_time_policy,
@@ -212,10 +193,7 @@ class ReadFromKafka(ExternalTransform):
start_read_time=start_read_time,
commit_offset_in_finalize=commit_offset_in_finalize,
timestamp_policy=timestamp_policy,
- consumer_polling_timeout=consumer_polling_timeout,
- redistribute=redistribute,
- redistribute_num_keys=redistribute_num_keys,
- allow_duplicates=allow_duplicates)),
+ consumer_polling_timeout=consumer_polling_timeout)),
expansion_service or default_io_expansion_service())
diff --git a/sdks/python/apache_beam/runners/portability/flink_runner_test.py b/sdks/python/apache_beam/runners/portability/flink_runner_test.py
index 4dc2446fdd9..f69ee1c24c4 100644
--- a/sdks/python/apache_beam/runners/portability/flink_runner_test.py
+++ b/sdks/python/apache_beam/runners/portability/flink_runner_test.py
@@ -252,9 +252,6 @@ class FlinkRunnerTest(portable_runner_test.PortableRunnerTest):
'LongDeserializer',
commit_offset_in_finalize=True,
timestamp_policy=ReadFromKafka.create_time_policy,
- redistribute=False,
- redistribute_num_keys=0,
- allow_duplicates=False,
expansion_service=self.get_expansion_service()))
self.assertTrue(
'No resolvable bootstrap urls given in bootstrap.servers' in str(