You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by ma...@apache.org on 2023/03/06 19:56:46 UTC
[flink-connector-kafka] 10/27: [hotfix][connectors][kafka] Remove unused classes and methods
This is an automated email from the ASF dual-hosted git repository.
martijnvisser pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/flink-connector-kafka.git
commit 7692937ec6254fee599c6cabaa99157f8e7ff08d
Author: Sergey Nuyanzin <sn...@gmail.com>
AuthorDate: Mon Oct 31 12:19:53 2022 +0100
[hotfix][connectors][kafka] Remove unused classes and methods
---
.../connector/kafka/sink/KafkaSinkITCase.java | 16 -----
.../source/enumerator/KafkaEnumeratorTest.java | 19 -----
.../connectors/kafka/KafkaProducerTestBase.java | 84 ----------------------
3 files changed, 119 deletions(-)
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
index a5e7f1f..942902f 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/sink/KafkaSinkITCase.java
@@ -67,11 +67,7 @@ import org.apache.kafka.clients.admin.AdminClient;
import org.apache.kafka.clients.admin.CreateTopicsResult;
import org.apache.kafka.clients.admin.DeleteTopicsResult;
import org.apache.kafka.clients.admin.NewTopic;
-import org.apache.kafka.clients.consumer.Consumer;
-import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerRecord;
-import org.apache.kafka.clients.consumer.KafkaConsumer;
-import org.apache.kafka.common.serialization.ByteArrayDeserializer;
import org.junit.After;
import org.junit.AfterClass;
import org.junit.Before;
@@ -426,18 +422,6 @@ public class KafkaSinkITCase extends TestLogger {
return standardProps;
}
- private static Consumer<byte[], byte[]> createTestConsumer(
- String topic, Properties properties) {
- final Properties consumerConfig = new Properties();
- consumerConfig.putAll(properties);
- consumerConfig.put("key.deserializer", ByteArrayDeserializer.class.getName());
- consumerConfig.put("value.deserializer", ByteArrayDeserializer.class.getName());
- consumerConfig.put(ConsumerConfig.ISOLATION_LEVEL_CONFIG, "read_committed");
- final KafkaConsumer<byte[], byte[]> kafkaConsumer = new KafkaConsumer<>(consumerConfig);
- kafkaConsumer.subscribe(Collections.singletonList(topic));
- return kafkaConsumer;
- }
-
private void createTestTopic(String topic, int numPartitions, short replicationFactor)
throws ExecutionException, InterruptedException, TimeoutException {
final CreateTopicsResult result =
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
index 307e353..8c39b54 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/connector/kafka/source/enumerator/KafkaEnumeratorTest.java
@@ -558,23 +558,4 @@ public class KafkaEnumeratorTest {
context.runNextOneTimeCallable();
}
}
-
- // -------------- private class ----------------
-
- private static class BlockingClosingContext
- extends MockSplitEnumeratorContext<KafkaPartitionSplit> {
-
- public BlockingClosingContext(int parallelism) {
- super(parallelism);
- }
-
- @Override
- public void close() {
- try {
- Thread.sleep(Long.MAX_VALUE);
- } catch (InterruptedException e) {
- // let it go.
- }
- }
- }
}
diff --git a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
index 8bcee36..6243810 100644
--- a/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
+++ b/flink-connector-kafka/src/test/java/org/apache/flink/streaming/connectors/kafka/KafkaProducerTestBase.java
@@ -23,15 +23,10 @@ import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.api.common.serialization.TypeInformationSerializationSchema;
-import org.apache.flink.api.common.state.CheckpointListener;
import org.apache.flink.api.common.typeinfo.BasicTypeInfo;
import org.apache.flink.api.common.typeinfo.TypeHint;
import org.apache.flink.api.common.typeinfo.TypeInformation;
import org.apache.flink.api.java.tuple.Tuple2;
-import org.apache.flink.configuration.Configuration;
-import org.apache.flink.runtime.state.FunctionInitializationContext;
-import org.apache.flink.runtime.state.FunctionSnapshotContext;
-import org.apache.flink.streaming.api.checkpoint.CheckpointedFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
@@ -63,8 +58,6 @@ import static org.assertj.core.api.Assertions.fail;
@SuppressWarnings("serial")
public abstract class KafkaProducerTestBase extends KafkaTestBaseWithFlink {
- private static final long KAFKA_READ_TIMEOUT = 60_000L;
-
/**
* This tests verifies that custom partitioning works correctly, with a default topic and
* dynamic topic. The number of partitions for each topic is deliberately different.
@@ -405,81 +398,4 @@ public abstract class KafkaProducerTestBase extends KafkaTestBaseWithFlink {
}
}
}
-
- private static class BrokerRestartingMapper<T> extends RichMapFunction<T, T>
- implements CheckpointedFunction, CheckpointListener {
-
- private static final long serialVersionUID = 6334389850158707313L;
-
- public static volatile boolean triggeredShutdown;
- public static volatile int lastSnapshotedElementBeforeShutdown;
- public static volatile Runnable shutdownAction;
-
- private final int failCount;
- private int numElementsTotal;
-
- private boolean failer;
-
- public static void resetState(Runnable shutdownAction) {
- triggeredShutdown = false;
- lastSnapshotedElementBeforeShutdown = 0;
- BrokerRestartingMapper.shutdownAction = shutdownAction;
- }
-
- public BrokerRestartingMapper(int failCount) {
- this.failCount = failCount;
- }
-
- @Override
- public void open(Configuration parameters) {
- failer = getRuntimeContext().getIndexOfThisSubtask() == 0;
- }
-
- @Override
- public T map(T value) throws Exception {
- numElementsTotal++;
- Thread.sleep(10);
-
- if (!triggeredShutdown && failer && numElementsTotal >= failCount) {
- // shut down a Kafka broker
- triggeredShutdown = true;
- shutdownAction.run();
- }
- return value;
- }
-
- @Override
- public void notifyCheckpointComplete(long checkpointId) {}
-
- @Override
- public void notifyCheckpointAborted(long checkpointId) {}
-
- @Override
- public void snapshotState(FunctionSnapshotContext context) throws Exception {
- if (!triggeredShutdown) {
- lastSnapshotedElementBeforeShutdown = numElementsTotal;
- }
- }
-
- @Override
- public void initializeState(FunctionInitializationContext context) throws Exception {}
- }
-
- private static final class InfiniteIntegerSource implements SourceFunction<Integer> {
-
- private volatile boolean running = true;
- private int counter = 0;
-
- @Override
- public void run(SourceContext<Integer> ctx) throws Exception {
- while (running) {
- ctx.collect(counter++);
- }
- }
-
- @Override
- public void cancel() {
- running = false;
- }
- }
}