You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@ignite.apache.org by tl...@apache.org on 2024/01/12 15:32:15 UTC
(ignite-extensions) branch master updated: IGNITE-18242 upgrade kafka dependency (#251)
This is an automated email from the ASF dual-hosted git repository.
tledkov pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/ignite-extensions.git
The following commit(s) were added to refs/heads/master by this push:
new bc8f250b IGNITE-18242 upgrade kafka dependency (#251)
bc8f250b is described below
commit bc8f250b891640b8f65eebc13673e643f8d029e3
Author: xintrian <38...@users.noreply.github.com>
AuthorDate: Fri Jan 12 18:32:08 2024 +0300
IGNITE-18242 upgrade kafka dependency (#251)
---
modules/cdc-ext/pom.xml | 10 +-
modules/kafka-ext/pom.xml | 4 +-
.../stream/kafka/KafkaIgniteStreamerSelfTest.java | 4 +-
.../ignite/stream/kafka/TestKafkaBroker.java | 116 ++++++++-------------
.../kafka/connect/IgniteSinkConnectorTest.java | 12 ++-
.../kafka/connect/IgniteSourceConnectorTest.java | 14 +--
6 files changed, 75 insertions(+), 85 deletions(-)
diff --git a/modules/cdc-ext/pom.xml b/modules/cdc-ext/pom.xml
index b1e63b5a..e9b4d298 100644
--- a/modules/cdc-ext/pom.xml
+++ b/modules/cdc-ext/pom.xml
@@ -32,7 +32,7 @@
</parent>
<properties>
- <kafka.version>2.8.2</kafka.version>
+ <kafka.version>3.4.0</kafka.version>
</properties>
<artifactId>ignite-cdc-ext</artifactId>
@@ -179,6 +179,14 @@
</execution>
</executions>
</plugin>
+ <plugin>
+ <groupId>org.apache.maven.plugins</groupId>
+ <artifactId>maven-compiler-plugin</artifactId>
+ <configuration>
+ <source>8</source>
+ <target>8</target>
+ </configuration>
+ </plugin>
</plugins>
</build>
</project>
diff --git a/modules/kafka-ext/pom.xml b/modules/kafka-ext/pom.xml
index a6e2fbd2..5aaacd3b 100644
--- a/modules/kafka-ext/pom.xml
+++ b/modules/kafka-ext/pom.xml
@@ -65,7 +65,7 @@
<dependency>
<groupId>org.apache.kafka</groupId>
- <artifactId>kafka_2.11</artifactId>
+ <artifactId>kafka_2.12</artifactId>
<version>${kafka.version}</version>
<scope>test</scope>
<exclusions>
@@ -169,7 +169,7 @@
<dependency>
<groupId>org.apache.kafka</groupId>
- <artifactId>kafka_2.11</artifactId>
+ <artifactId>kafka_2.12</artifactId>
<version>${kafka.version}</version>
<classifier>test</classifier>
<scope>test</scope>
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
index 484608d1..ef4256e8 100644
--- a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/KafkaIgniteStreamerSelfTest.java
@@ -17,6 +17,7 @@
package org.apache.ignite.stream.kafka;
+import java.io.IOException;
import java.util.ArrayList;
import java.util.Arrays;
import java.util.Collections;
@@ -26,6 +27,7 @@ import java.util.Map;
import java.util.Properties;
import java.util.UUID;
import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import org.apache.ignite.Ignite;
import org.apache.ignite.IgniteCache;
@@ -105,7 +107,7 @@ public class KafkaIgniteStreamerSelfTest extends GridCommonAbstractTest {
* @throws InterruptedException If interrupted.
*/
@Test
- public void testKafkaStreamer() throws TimeoutException, InterruptedException {
+ public void testKafkaStreamer() throws TimeoutException, InterruptedException, IOException, ExecutionException {
embeddedBroker.createTopic(TOPIC_NAME, PARTITIONS, REPLICATION_FACTOR);
Map<String, String> keyValMap = produceStream(TOPIC_NAME);
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java
index 24dbfc8e..7d45f29b 100644
--- a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/TestKafkaBroker.java
@@ -21,26 +21,24 @@ import java.io.File;
import java.io.IOException;
import java.nio.file.Files;
import java.nio.file.Path;
-import java.util.ArrayList;
+import java.util.Collections;
import java.util.List;
import java.util.Properties;
+import java.util.concurrent.ExecutionException;
import java.util.concurrent.TimeoutException;
import kafka.server.KafkaConfig;
import kafka.server.KafkaServer;
import kafka.utils.TestUtils;
-import kafka.utils.ZkUtils;
-import kafka.zk.KafkaZkClient;
-import org.I0Itec.zkclient.ZkClient;
-import org.I0Itec.zkclient.ZkConnection;
import org.apache.curator.test.TestingServer;
import org.apache.ignite.internal.util.typedef.internal.U;
+import org.apache.kafka.clients.admin.AdminClient;
+import org.apache.kafka.clients.admin.NewTopic;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.Producer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.serialization.StringSerializer;
import org.apache.kafka.common.utils.SystemTime;
-import scala.Tuple2;
/**
* Kafka Test Broker.
@@ -62,31 +60,44 @@ public class TestKafkaBroker {
private static final int BROKER_PORT = 11092;
/** Kafka config. */
- private KafkaConfig kafkaCfg;
+ private final KafkaConfig kafkaCfg;
/** Kafka server. */
- private KafkaServer kafkaSrv;
+ private final KafkaServer kafkaSrv;
/** ZooKeeper. */
- private TestingServer zkServer;
+ private final TestingServer zkServer;
- /** Kafka Zookeeper utils. */
- private ZkUtils zkUtils;
+ /** Kafka Admin Client. */
+ private AdminClient admin;
/**
* Kafka broker constructor.
*/
public TestKafkaBroker() {
try {
- setupZooKeeper();
+ zkServer = new TestingServer(ZK_PORT, true);
+ kafkaCfg = new KafkaConfig(getKafkaConfig());
+ kafkaSrv = TestUtils.createServer(kafkaCfg, new SystemTime());
- setupKafkaServer();
+ kafkaSrv.startup();
}
catch (Exception e) {
throw new RuntimeException("Failed to start Kafka: " + e);
}
}
+ /**
+ * @return Existing admin client or creates new
+ * @throws IOException
+ */
+ public synchronized AdminClient admin() throws IOException {
+ if (admin == null)
+ admin = AdminClient.create(getKafkaConfig());
+
+ return admin;
+ }
+
/**
* Creates a topic.
*
@@ -97,50 +108,40 @@ public class TestKafkaBroker {
* @throws InterruptedException If interrupted.
*/
public void createTopic(String topic, int partitions, int replicationFactor)
- throws TimeoutException, InterruptedException {
- List<KafkaServer> servers = new ArrayList<>();
-
- servers.add(kafkaSrv);
-
- KafkaZkClient client = kafkaSrv.zkClient();
-
- TestUtils.createTopic(client, topic, partitions, replicationFactor,
- scala.collection.JavaConversions.asScalaBuffer(servers), new Properties());
+ throws IOException, ExecutionException, InterruptedException {
+ admin().createTopics(Collections.singleton(new NewTopic(topic, partitions, (short)replicationFactor)))
+ .all().get();
}
/**
* Sends a message to Kafka broker.
*
* @param records List of records.
- * @return Producer used to send the message.
*/
public void sendMessages(List<ProducerRecord<String, String>> records) {
- Producer<String, String> producer = new KafkaProducer<>(getProducerConfig());
-
- for (ProducerRecord<String, String> rec : records)
- producer.send(rec);
-
- producer.flush();
- producer.close();
+ try (Producer<String, String> producer = new KafkaProducer<>(getProducerConfig())) {
+ for (ProducerRecord<String, String> rec : records)
+ producer.send(rec);
+ }
}
/**
* Shuts down test Kafka broker.
*/
public void shutdown() {
- if (zkUtils != null)
- zkUtils.close();
-
- if (kafkaSrv != null)
- kafkaSrv.shutdown();
-
- if (zkServer != null) {
- try {
- zkServer.stop();
- }
- catch (IOException ignored) {
- // No-op.
- }
+ if (admin != null)
+ admin.close();
+
+ assert kafkaSrv != null;
+ assert zkServer != null;
+
+ kafkaSrv.shutdown();
+
+ try {
+ zkServer.stop();
+ }
+ catch (IOException ignored) {
+ // No-op.
}
List<String> logDirs = scala.collection.JavaConversions.seqAsJavaList(kafkaCfg.logDirs());
@@ -149,33 +150,6 @@ public class TestKafkaBroker {
U.delete(new File(logDir));
}
- /**
- * Sets up test Kafka broker.
- *
- * @throws IOException If failed.
- */
- private void setupKafkaServer() throws IOException {
- kafkaCfg = new KafkaConfig(getKafkaConfig());
-
- kafkaSrv = TestUtils.createServer(kafkaCfg, new SystemTime());
-
- kafkaSrv.startup();
- }
-
- /**
- * Sets up ZooKeeper test server.
- *
- * @throws Exception If failed.
- */
- private void setupZooKeeper() throws Exception {
- zkServer = new TestingServer(ZK_PORT, true);
-
- Tuple2<ZkClient, ZkConnection> zkTuple = ZkUtils.createZkClientAndConnection(zkServer.getConnectString(),
- ZK_SESSION_TIMEOUT, ZK_CONNECTION_TIMEOUT);
-
- zkUtils = new ZkUtils(zkTuple._1(), zkTuple._2(), false);
- }
-
/**
* Obtains Kafka config.
*
@@ -189,10 +163,12 @@ public class TestKafkaBroker {
props.put("zookeeper.connect", zkServer.getConnectString());
props.put("host.name", BROKER_HOST);
props.put("port", BROKER_PORT);
+ props.put("bootstrap.servers", getBrokerAddress());
props.put("offsets.topic.replication.factor", "1");
props.put("log.dir", createTmpDir("_cfg").getAbsolutePath());
props.put("log.flush.interval.messages", "1");
props.put("log.flush.interval.ms", "10");
+ props.put("listeners", "PLAINTEXT://" + getBrokerAddress());
return props;
}
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java
index 8a70636b..430da505 100644
--- a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSinkConnectorTest.java
@@ -37,6 +37,7 @@ import org.apache.ignite.testframework.junits.common.GridCommonAbstractTest;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.Herder;
import org.apache.kafka.connect.runtime.Worker;
@@ -49,7 +50,6 @@ import org.apache.kafka.connect.sink.SinkConnector;
import org.apache.kafka.connect.sink.SinkRecord;
import org.apache.kafka.connect.storage.OffsetBackingStore;
import org.apache.kafka.connect.util.Callback;
-import org.apache.kafka.connect.util.ConnectUtils;
import org.apache.kafka.connect.util.FutureCallback;
import org.junit.Test;
@@ -103,10 +103,14 @@ public class IgniteSinkConnectorTest extends GridCommonAbstractTest {
OffsetBackingStore offBackingStore = mock(OffsetBackingStore.class);
offBackingStore.configure(workerCfg);
- worker = new Worker(WORKER_ID, new SystemTime(), new Plugins(props), workerCfg, offBackingStore);
+ AllConnectorClientConfigOverridePolicy allConnectorClientCfgOverridePlc
+ = new AllConnectorClientConfigOverridePolicy();
+
+ worker = new Worker(WORKER_ID, new SystemTime(), new Plugins(props), workerCfg, offBackingStore,
+ allConnectorClientCfgOverridePlc);
worker.start();
- herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg));
+ herder = new StandaloneHerder(worker, workerCfg.kafkaClusterId(), allConnectorClientCfgOverridePlc);
herder.start();
}
@@ -278,8 +282,6 @@ public class IgniteSinkConnectorTest extends GridCommonAbstractTest {
private Map<String, String> makeWorkerProps() {
Map<String, String> props = new HashMap<>();
- props.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
- props.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
props.put("internal.key.converter.schemas.enable", "false");
props.put("internal.value.converter.schemas.enable", "false");
props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
diff --git a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
index a1109e76..afd9e348 100644
--- a/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
+++ b/modules/kafka-ext/src/test/java/org/apache/ignite/stream/kafka/connect/IgniteSourceConnectorTest.java
@@ -43,6 +43,7 @@ import org.apache.kafka.clients.consumer.KafkaConsumer;
import org.apache.kafka.common.errors.WakeupException;
import org.apache.kafka.common.utils.SystemTime;
import org.apache.kafka.common.utils.Utils;
+import org.apache.kafka.connect.connector.policy.AllConnectorClientConfigOverridePolicy;
import org.apache.kafka.connect.runtime.ConnectorConfig;
import org.apache.kafka.connect.runtime.Herder;
import org.apache.kafka.connect.runtime.Worker;
@@ -53,7 +54,6 @@ import org.apache.kafka.connect.runtime.standalone.StandaloneConfig;
import org.apache.kafka.connect.runtime.standalone.StandaloneHerder;
import org.apache.kafka.connect.storage.MemoryOffsetBackingStore;
import org.apache.kafka.connect.util.Callback;
-import org.apache.kafka.connect.util.ConnectUtils;
import org.apache.kafka.connect.util.FutureCallback;
import org.junit.Test;
@@ -106,10 +106,14 @@ public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
MemoryOffsetBackingStore offBackingStore = new MemoryOffsetBackingStore();
offBackingStore.configure(workerCfg);
- worker = new Worker(WORKER_ID, new SystemTime(), new Plugins(props), workerCfg, offBackingStore);
+ AllConnectorClientConfigOverridePolicy allConnectorClientCfgOverridePlc
+ = new AllConnectorClientConfigOverridePolicy();
+
+ worker = new Worker(WORKER_ID, new SystemTime(), new Plugins(props), workerCfg, offBackingStore,
+ allConnectorClientCfgOverridePlc);
worker.start();
- herder = new StandaloneHerder(worker, ConnectUtils.lookupKafkaClusterId(workerCfg));
+ herder = new StandaloneHerder(worker, workerCfg.kafkaClusterId(), allConnectorClientCfgOverridePlc);
herder.start();
}
@@ -334,11 +338,9 @@ public class IgniteSourceConnectorTest extends GridCommonAbstractTest {
* @return Worker configurations.
* @throws IOException If failed.
*/
- private Map<String, String> makeWorkerProps() throws IOException {
+ private Map<String, String> makeWorkerProps(){
Map<String, String> props = new HashMap<>();
- props.put(WorkerConfig.INTERNAL_KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
- props.put(WorkerConfig.INTERNAL_VALUE_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");
props.put("internal.key.converter.schemas.enable", "false");
props.put("internal.value.converter.schemas.enable", "false");
props.put(WorkerConfig.KEY_CONVERTER_CLASS_CONFIG, "org.apache.kafka.connect.storage.StringConverter");