This is an automated email from the ASF dual-hosted git repository.
sarathsubramanian pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/atlas.git
The following commit(s) were added to refs/heads/master by this push:
new 4bec89039 ATLAS-5336: Upgrade Kafka to 3.9.1; embedded notification
broker uses KRaft (no ZooKeeper) (#689)
4bec89039 is described below
commit 4bec890393d06abea98ee135472b4df9f25d0d87
Author: Ramachandran Krishnan <[email protected]>
AuthorDate: Wed Jul 8 09:55:12 2026 +0530
ATLAS-5336: Upgrade Kafka to 3.9.1; embedded notification broker uses KRaft
(no ZooKeeper) (#689)
* ATLAS-5336: Upgrade Kafka to 3.9.1; embedded notification broker uses
KRaft
Bump kafka-clients and embedded test broker from 2.8.2 to 3.9.1 (Scala
2.13),
replace ZooKeeper-based EmbeddedKafkaServer with KafkaClusterTestKit, and
update docker dev Kafka image packaging to kafka_2.13-3.9.1.
* Document jackson-jaxrs 2.12.7 pin for Jersey 1.19 compatibility.
Point readers to ATLAS-5190 for full Jackson unification via Jersey 2.x.
* Remove unused junit-jupiter-api from atlas-notification compile deps.
It leaked to falcon-bridge via atlas-notification and caused Failsafe to
select JUnit Platform, which then failed on transitive junit 3.8.1 from
falcon-common during FalconHookIT discovery.
* Scope Kafka test and unit test deps in notification to test.
Kafka 3.9 TestUtils needs junit-jupiter-api at test runtime; keep it off the
compile classpath so falcon-bridge Failsafe stays on TestNG. Align mockito
and
testng with other Atlas modules (test scope).
* Add test-scoped junit-jupiter-api to webapp for Kafka embedded broker
tests.
NotificationHookConsumerKafkaTest starts EmbeddedKafkaServer, which uses
Kafka
3.9 TestUtils requiring JUnit Jupiter on the test classpath. Test-scoped
deps
from atlas-notification do not propagate to webapp.
* Fix flaky ImportTaskListenerImplTest async consumer assertions on CI.
Shut down the import executor in @AfterMethod and wait with CountDownLatch
instead of Thread.sleep so Mockito state does not leak between tests.
* Harden BasicSearchIT and AdminExportImportTestIT against CI flakes.
Scope hive searches to largedb. entities from the imported test dataset
instead of the shared @cl1 cluster suffix, and tolerate Stream Closed
during AdminExportImportTestIT teardown when the embedded server stops.
* Fix IT scoping and Failsafe/TestNG classpath for Kafka 3.9.
Scope BasicSearchIT to default. entities from hive-db-50-tables.zip instead
of the shared @cl1 suffix, skip scoping for fulltext and paginated searches,
keep junit-jupiter on the notification compile classpath for embedded KRaft
runtime, and pin Failsafe to TestNG after junit landed on the webapp test
path.
* Bind embedded KRaft broker to configured bootstrap.servers for ITs.
Derive EXTERNAL and CONTROLLER listeners from atlas.kafka.bootstrap.servers
so Jetty and Failsafe JVMs share the same broker port, fixing
NotificationHookConsumerIT topic timeouts on CI.
* Pre-serialize classification request bodies to avoid Codehaus Jackson
conflicts.
HiveHookIT fails when Jersey serializes AtlasClassification because
isPropagate()
and getPropagate() both map to "propagate"; AtlasType.toJson() matches the
entity fix.
* Pin Jackson 2.16 on sqoop-bridge IT classpath for AtlasClient v1.
SqoopHookIT fails with NoClassDefFoundError for StreamConstraintsException
when jackson-databind 2.16 is paired with an older jackson-core from
Hadoop/Hive.
* Pin Jackson 2.16 databind on storm-bridge IT classpath for AtlasClient v1.
StormAtlasHookIT fails with NoSuchFieldError for
READ_UNKNOWN_ENUM_VALUES_USING_DEFAULT_VALUE
when jackson-databind 2.16 is mixed with an older databind from storm-core
on the test classpath.
Co-authored-by: Cursor <[email protected]>
* Skip trino-extractor Failsafe ITs on default mvn verify.
Pinning Failsafe to TestNG for the Kafka upgrade requires testng on the
module test classpath; the extractor only has a stub IT and no test deps
yet.
---------
Co-authored-by: ramk <[email protected]>
Co-authored-by: Cursor <[email protected]>
---
addons/sqoop-bridge/pom.xml | 13 ++
addons/storm-bridge/pom.xml | 8 +-
addons/trino-extractor/pom.xml | 2 +
.../main/java/org/apache/atlas/AtlasClientV2.java | 8 +-
dev-support/atlas-docker/.env | 2 +-
dev-support/atlas-docker/Dockerfile.atlas-kafka | 9 +-
dev-support/atlas-docker/download-archives.sh | 2 +-
distro/src/conf/atlas-application.properties | 5 +-
notification/pom.xml | 57 +++++-
.../apache/atlas/kafka/EmbeddedKafkaServer.java | 212 +++++++++++++--------
.../org/apache/atlas/kafka/KafkaNotification.java | 11 ++
pom.xml | 15 +-
webapp/pom.xml | 33 +++-
.../notification/ImportTaskListenerImplTest.java | 52 +++--
.../atlas/web/integration/BasicSearchIT.java | 6 +-
.../web/resources/AdminExportImportTestIT.java | 22 +++
16 files changed, 331 insertions(+), 126 deletions(-)
diff --git a/addons/sqoop-bridge/pom.xml b/addons/sqoop-bridge/pom.xml
index ea1bc94ba..f4954624c 100644
--- a/addons/sqoop-bridge/pom.xml
+++ b/addons/sqoop-bridge/pom.xml
@@ -34,6 +34,12 @@
<dependencies>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.databind.version}</version>
+ </dependency>
+
<dependency>
<groupId>org.apache.atlas</groupId>
<artifactId>atlas-client-v1</artifactId>
@@ -129,6 +135,13 @@
</exclusions>
</dependency>
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-core</artifactId>
+ <version>${jackson.version}</version>
+ <scope>test</scope>
+ </dependency>
+
<dependency>
<groupId>org.apache.atlas</groupId>
<artifactId>atlas-graphdb-impls</artifactId>
diff --git a/addons/storm-bridge/pom.xml b/addons/storm-bridge/pom.xml
index 42e325f1c..1033a3739 100644
--- a/addons/storm-bridge/pom.xml
+++ b/addons/storm-bridge/pom.xml
@@ -33,6 +33,12 @@
<description>Apache Atlas Storm Bridge Module</description>
<dependencies>
+ <!-- apache atlas core dependencies -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.core</groupId>
+ <artifactId>jackson-databind</artifactId>
+ <version>${jackson.databind.version}</version>
+ </dependency>
<dependency>
<groupId>commons-collections</groupId>
@@ -55,7 +61,7 @@
<artifactId>metrics-graphite</artifactId>
<version>${dropwizard-metrics}</version>
</dependency>
- <!-- apache atlas core dependencies -->
+
<dependency>
<groupId>org.apache.atlas</groupId>
<artifactId>atlas-client-v1</artifactId>
diff --git a/addons/trino-extractor/pom.xml b/addons/trino-extractor/pom.xml
index b66b957ca..9a15f3edf 100644
--- a/addons/trino-extractor/pom.xml
+++ b/addons/trino-extractor/pom.xml
@@ -35,6 +35,8 @@
<properties>
<checkstyle.failOnViolation>true</checkstyle.failOnViolation>
<checkstyle.skip>false</checkstyle.skip>
+ <!-- No live-stack ITs yet; avoid Failsafe/TestNG on default mvn
verify (see ATLAS-5338). -->
+ <skipITs>true</skipITs>
</properties>
<dependencies>
diff --git a/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
b/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
index daa600949..c7260ff6c 100644
--- a/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
+++ b/client/client-v2/src/main/java/org/apache/atlas/AtlasClientV2.java
@@ -518,23 +518,23 @@ public class AtlasClientV2 extends AtlasBaseClient {
}
public void addClassifications(String guid, List<AtlasClassification>
classifications) throws AtlasServiceException {
- callAPI(formatPathParameters(API_V2.ADD_CLASSIFICATIONS, guid),
(Class<?>) null, classifications, (String[]) null);
+ callAPI(formatPathParameters(API_V2.ADD_CLASSIFICATIONS, guid),
(Class<?>) null, AtlasType.toJson(classifications), (String[]) null);
}
public void addClassifications(String typeName, Map<String, String>
uniqAttributes, List<AtlasClassification> classifications) throws
AtlasServiceException {
MultivaluedMap<String, String> queryParams =
attributesToQueryParams(uniqAttributes);
-
callAPI(formatPathParameters(API_V2.ADD_CLASSIFICATION_BY_TYPE_AND_ATTRIBUTE,
typeName), (Class<?>) null, classifications, queryParams);
+
callAPI(formatPathParameters(API_V2.ADD_CLASSIFICATION_BY_TYPE_AND_ATTRIBUTE,
typeName), (Class<?>) null, AtlasType.toJson(classifications), queryParams);
}
public void updateClassifications(String guid, List<AtlasClassification>
classifications) throws AtlasServiceException {
- callAPI(formatPathParameters(API_V2.UPDATE_CLASSIFICATIONS, guid),
(Class<?>) null, classifications);
+ callAPI(formatPathParameters(API_V2.UPDATE_CLASSIFICATIONS, guid),
(Class<?>) null, AtlasType.toJson(classifications));
}
public void updateClassifications(String typeName, Map<String, String>
uniqAttributes, List<AtlasClassification> classifications) throws
AtlasServiceException {
MultivaluedMap<String, String> queryParams =
attributesToQueryParams(uniqAttributes);
-
callAPI(formatPathParameters(API_V2.UPDATE_CLASSIFICATION_BY_TYPE_AND_ATTRIBUTE,
typeName), (Class<?>) null, classifications, queryParams);
+
callAPI(formatPathParameters(API_V2.UPDATE_CLASSIFICATION_BY_TYPE_AND_ATTRIBUTE,
typeName), (Class<?>) null, AtlasType.toJson(classifications), queryParams);
}
public String setClassifications(AtlasEntityHeaders entityHeaders) throws
AtlasServiceException {
diff --git a/dev-support/atlas-docker/.env b/dev-support/atlas-docker/.env
index c07b9fb63..2041f68aa 100644
--- a/dev-support/atlas-docker/.env
+++ b/dev-support/atlas-docker/.env
@@ -21,7 +21,7 @@ ATLAS_VERSION=3.0.0-SNAPSHOT
UBUNTU_VERSION=20.04
HADOOP_VERSION=3.4.2
HBASE_VERSION=2.6.4
-KAFKA_VERSION=2.8.2
+KAFKA_VERSION=3.9.1
HIVE_VERSION=4.1.0
ATLAS_BACKEND=hbase
diff --git a/dev-support/atlas-docker/Dockerfile.atlas-kafka
b/dev-support/atlas-docker/Dockerfile.atlas-kafka
index 5c5dba802..4406fbd90 100644
--- a/dev-support/atlas-docker/Dockerfile.atlas-kafka
+++ b/dev-support/atlas-docker/Dockerfile.atlas-kafka
@@ -18,17 +18,18 @@ FROM atlas-base:latest
ARG ATLAS_VERSION
ARG KAFKA_VERSION
+ARG KAFKA_SCALA_VERSION=2.13
COPY ./dist/apache-atlas-${ATLAS_VERSION}-kafka-hook.tar.gz /home/atlas/dist/
-COPY ./downloads/kafka_2.12-${KAFKA_VERSION}.tgz /home/atlas/dist/
+COPY ./downloads/kafka_${KAFKA_SCALA_VERSION}-${KAFKA_VERSION}.tgz
/home/atlas/dist/
COPY ./scripts/atlas-kafka-setup.sh /home/atlas/scripts/
COPY ./scripts/atlas-kafka.sh /home/atlas/scripts/
COPY ./scripts/atlas-kafka-application.properties /home/atlas/scripts/
-RUN tar xvfz /home/atlas/dist/kafka_2.12-${KAFKA_VERSION}.tgz
--directory=/opt/ && \
- ln -s /opt/kafka_2.12-${KAFKA_VERSION} /opt/kafka && \
- rm -f /home/atlas/dist/kafka_2.12-${KAFKA_VERSION}.tgz && \
+RUN tar xvfz
/home/atlas/dist/kafka_${KAFKA_SCALA_VERSION}-${KAFKA_VERSION}.tgz
--directory=/opt/ && \
+ ln -s /opt/kafka_${KAFKA_SCALA_VERSION}-${KAFKA_VERSION} /opt/kafka && \
+ rm -f /home/atlas/dist/kafka_${KAFKA_SCALA_VERSION}-${KAFKA_VERSION}.tgz
&& \
tar xvfz /home/atlas/dist/apache-atlas-${ATLAS_VERSION}-kafka-hook.tar.gz
--directory=/opt/ && \
ln -s /opt/apache-atlas-kafka-hook-${ATLAS_VERSION}
/opt/apache-atlas-kafka-hook && \
rm -f /home/atlas/dist/apache-atlas-${ATLAS_VERSION}-kafka-hook.tar.gz && \
diff --git a/dev-support/atlas-docker/download-archives.sh
b/dev-support/atlas-docker/download-archives.sh
index 5c22fe73b..17ea80aff 100755
--- a/dev-support/atlas-docker/download-archives.sh
+++ b/dev-support/atlas-docker/download-archives.sh
@@ -46,6 +46,6 @@ downloadIfNotPresent() {
downloadIfNotPresent hadoop-${HADOOP_VERSION}.tar.gz
https://archive.apache.org/dist/hadoop/common/hadoop-${HADOOP_VERSION}
downloadIfNotPresent hbase-${HBASE_VERSION}-bin.tar.gz
https://archive.apache.org/dist/hbase/${HBASE_VERSION}
-downloadIfNotPresent kafka_2.12-${KAFKA_VERSION}.tgz
https://archive.apache.org/dist/kafka/${KAFKA_VERSION}
+downloadIfNotPresent kafka_2.13-${KAFKA_VERSION}.tgz
https://archive.apache.org/dist/kafka/${KAFKA_VERSION}
downloadIfNotPresent apache-hive-${HIVE_VERSION}-bin.tar.gz
https://archive.apache.org/dist/hive/hive-${HIVE_VERSION}
diff --git a/distro/src/conf/atlas-application.properties
b/distro/src/conf/atlas-application.properties
index 126143a20..35ba5a3db 100755
--- a/distro/src/conf/atlas-application.properties
+++ b/distro/src/conf/atlas-application.properties
@@ -90,11 +90,8 @@ atlas.graph.index.search.max-result-set-size=150
######### Notification Configs #########
atlas.notification.embedded=true
atlas.kafka.data=${sys:atlas.home}/data/kafka
-atlas.kafka.zookeeper.connect=localhost:9026
+# Embedded broker uses KRaft (no ZooKeeper). bootstrap.servers is updated at
runtime when embedded=true.
atlas.kafka.bootstrap.servers=localhost:9027
-atlas.kafka.zookeeper.session.timeout.ms=400
-atlas.kafka.zookeeper.connection.timeout.ms=200
-atlas.kafka.zookeeper.sync.time.ms=20
atlas.kafka.auto.commit.interval.ms=1000
atlas.kafka.hook.group.id=atlas
diff --git a/notification/pom.xml b/notification/pom.xml
index 1885f7c8d..22c1f6021 100644
--- a/notification/pom.xml
+++ b/notification/pom.xml
@@ -63,21 +63,49 @@
<groupId>org.apache.kafka</groupId>
<artifactId>kafka-clients</artifactId>
</dependency>
-
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-clients</artifactId>
+ <version>${kafka.version}</version>
+ <classifier>test</classifier>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-metadata</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-server</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-server-common</artifactId>
+ <version>${kafka.version}</version>
+ </dependency>
+ <dependency>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka-server-common</artifactId>
+ <version>${kafka.version}</version>
+ <classifier>test</classifier>
+ </dependency>
<dependency>
<groupId>org.apache.kafka</groupId>
<artifactId>kafka_${kafka.scala.binary.version}</artifactId>
</dependency>
-
<dependency>
- <groupId>org.mockito</groupId>
- <artifactId>mockito-core</artifactId>
- <version>${mockito.version}</version>
+ <groupId>org.apache.kafka</groupId>
+ <artifactId>kafka_${kafka.scala.binary.version}</artifactId>
+ <version>${kafka.version}</version>
+ <classifier>test</classifier>
</dependency>
+ <!-- KafkaClusterTestKit (embedded KRaft broker) calls
org.apache.kafka.test.TestUtils, which uses junit.jupiter at runtime in the
WAR. -->
<dependency>
- <groupId>org.testng</groupId>
- <artifactId>testng</artifactId>
+ <groupId>org.junit.jupiter</groupId>
+ <artifactId>junit-jupiter-api</artifactId>
+ <version>5.11.4</version>
</dependency>
<dependency>
@@ -94,6 +122,19 @@
<scope>test</scope>
</dependency>
+ <dependency>
+ <groupId>org.mockito</groupId>
+ <artifactId>mockito-core</artifactId>
+ <version>${mockito.version}</version>
+ <scope>test</scope>
+ </dependency>
+
+ <dependency>
+ <groupId>org.testng</groupId>
+ <artifactId>testng</artifactId>
+ <scope>test</scope>
+ </dependency>
+
</dependencies>
<build>
@@ -172,7 +213,7 @@
<artifactItem>
<groupId>org.apache.zookeeper</groupId>
<artifactId>zookeeper</artifactId>
- <version>3.4.6</version>
+ <version>${zookeeper.version}</version>
</artifactItem>
</artifactItems>
</configuration>
diff --git
a/notification/src/main/java/org/apache/atlas/kafka/EmbeddedKafkaServer.java
b/notification/src/main/java/org/apache/atlas/kafka/EmbeddedKafkaServer.java
index 23158f521..654caa7d9 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/EmbeddedKafkaServer.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/EmbeddedKafkaServer.java
@@ -17,36 +17,29 @@
*/
package org.apache.atlas.kafka;
-import kafka.server.KafkaConfig;
-import kafka.server.KafkaServer;
-import kafka.zookeeper.ZooKeeperClientException;
+import kafka.testkit.KafkaClusterTestKit;
+import kafka.testkit.TestKitNodes;
import org.apache.atlas.ApplicationProperties;
import org.apache.atlas.AtlasException;
import org.apache.atlas.service.Service;
import org.apache.atlas.util.CommandHandlerUtility;
import org.apache.commons.configuration2.Configuration;
import org.apache.commons.configuration2.ConfigurationConverter;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.kafka.clients.CommonClientConfigs;
import org.apache.kafka.clients.producer.ProducerConfig;
-import org.apache.kafka.common.KafkaException;
-import org.apache.kafka.common.utils.Time;
-import org.apache.zookeeper.server.NIOServerCnxnFactory;
-import org.apache.zookeeper.server.ServerCnxnFactory;
-import org.apache.zookeeper.server.ZooKeeperServer;
+import org.apache.kafka.common.utils.Exit;
+import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.core.annotation.Order;
import org.springframework.stereotype.Component;
-import scala.Option;
import javax.inject.Inject;
import java.io.File;
-import java.io.IOException;
-import java.net.BindException;
-import java.net.InetSocketAddress;
-import java.net.MalformedURLException;
-import java.net.URL;
import java.util.Properties;
+import java.util.concurrent.atomic.AtomicReference;
@Component
@Order(3)
@@ -59,14 +52,16 @@ public class EmbeddedKafkaServer implements Service {
private static final String ATLAS_KAFKA_DATA = "data";
private static final int MAX_RETRY_TO_ACQUIRE_PORT = 3;
- private final boolean isEmbedded;
- private final Properties properties;
- private KafkaServer kafkaServer;
- private ServerCnxnFactory factory;
+ private final boolean isEmbedded;
+ private final Configuration applicationProperties;
+ private final Properties properties;
+
+ private KafkaClusterTestKit cluster;
@Inject
public EmbeddedKafkaServer(Configuration applicationProperties) {
- Configuration kafkaConf =
ApplicationProperties.getSubsetConfiguration(applicationProperties,
PROPERTY_PREFIX);
+ this.applicationProperties = applicationProperties;
+ Configuration kafkaConf =
ApplicationProperties.getSubsetConfiguration(applicationProperties,
PROPERTY_PREFIX);
this.isEmbedded = applicationProperties.getBoolean(PROPERTY_EMBEDDED,
false);
this.properties = ConfigurationConverter.getProperties(kafkaConf);
@@ -78,8 +73,7 @@ public class EmbeddedKafkaServer implements Service {
if (isEmbedded) {
try {
- startZk();
- startKafka();
+ startKraftBroker();
} catch (Exception e) {
throw new AtlasException("Failed to start embedded kafka", e);
}
@@ -94,90 +88,152 @@ public class EmbeddedKafkaServer implements Service {
public void stop() {
LOG.info("==> EmbeddedKafkaServer.stop(isEmbedded={})", isEmbedded);
- if (kafkaServer != null) {
- kafkaServer.shutdown();
- }
-
- if (factory != null) {
- factory.shutdown();
- }
+ shutdownClusterQuietly();
LOG.info("<== EmbeddedKafka.stop(isEmbedded={})", isEmbedded);
}
- private String startZk() throws IOException, InterruptedException {
- String zkValue = properties.getProperty("zookeeper.connect");
+ private void startKraftBroker() throws Exception {
+ overrideExitMethods();
- LOG.info("Starting zookeeper at {}", zkValue);
+ File logDir = constructDir("kafka");
+ String configuredBootstrap =
properties.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
- URL zkAddress = getURL(zkValue);
- File snapshotDir = constructDir("zk/txn");
- File logDir = constructDir("zk/snap");
+ LOG.info("Starting embedded KRaft kafka (log.dir={},
bootstrap.servers={})", logDir.getAbsolutePath(), configuredBootstrap);
- for (int attemptCount = 0; attemptCount < MAX_RETRY_TO_ACQUIRE_PORT;
attemptCount++) {
+ for (int attempt = 0; attempt < MAX_RETRY_TO_ACQUIRE_PORT; attempt++) {
try {
- factory = NIOServerCnxnFactory.createFactory(new
InetSocketAddress(zkAddress.getHost(), zkAddress.getPort()), 1024);
- break;
- } catch (BindException e) {
- LOG.warn("Attempt {}: Starting zookeeper at {} failed",
attemptCount, zkValue);
+ startKraftBrokerOnce(logDir, configuredBootstrap);
+ return;
+ } catch (Exception e) {
+ LOG.warn("Attempt {}: failed to start embedded KRaft kafka",
attempt, e);
+
+ shutdownClusterQuietly();
- if (attemptCount == MAX_RETRY_TO_ACQUIRE_PORT - 1) {
+ if (attempt == MAX_RETRY_TO_ACQUIRE_PORT - 1) {
throw e;
}
-
CommandHandlerUtility.tryKillingProcessUsingPort(zkAddress.getPort(),
attemptCount != 0);
+ int port = parsePort(configuredBootstrap);
+
+ if (port > 0) {
+ CommandHandlerUtility.tryKillingProcessUsingPort(port,
attempt != 0);
+ }
}
}
+ }
+
+ private void startKraftBrokerOnce(File logDir, String configuredBootstrap)
throws Exception {
+ KafkaClusterTestKit.Builder clusterBuilder = new
KafkaClusterTestKit.Builder(
+ new TestKitNodes.Builder()
+ .setCombined(true)
+ .setNumBrokerNodes(1)
+ .setNumControllerNodes(1)
+ .build());
+
+ Properties brokerConfig = buildBrokerConfig(logDir,
configuredBootstrap);
- factory.startup(new ZooKeeperServer(snapshotDir, logDir, 500));
+ brokerConfig.forEach((key, value) ->
clusterBuilder.setConfigProp(key.toString(), value));
- String ret = factory.getLocalAddress().getAddress().toString();
+ cluster = clusterBuilder.build();
+ cluster.format();
+ cluster.startup();
+ cluster.waitForReadyBrokers();
- LOG.info("Embedded zookeeper for Kafka started at {}", ret);
+ String bootstrapServers = StringUtils.isNotEmpty(configuredBootstrap)
+ ? configuredBootstrap
+ :
cluster.clientProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG).toString();
- return ret;
+ LOG.info("Embedded KRaft kafka server started at {}",
bootstrapServers);
+
+ applicationProperties.setProperty(PROPERTY_PREFIX +
".bootstrap.servers", bootstrapServers);
+ properties.setProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG,
bootstrapServers);
}
- private void startKafka() throws IOException {
- String kafkaValue =
properties.getProperty(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG);
+ private Properties buildBrokerConfig(File logDir, String
configuredBootstrap) {
+ Properties brokerConfig = new Properties();
- LOG.info("Starting kafka at {}", kafkaValue);
+ brokerConfig.setProperty("log.dir", logDir.getAbsolutePath());
+ brokerConfig.setProperty("delete.topic.enable", "true");
+ brokerConfig.setProperty("auto.create.topics.enable", "true");
+ brokerConfig.setProperty("group.initial.rebalance.delay.ms", "0");
+ brokerConfig.setProperty("offsets.topic.replication.factor", "1");
+ brokerConfig.setProperty("transaction.state.log.replication.factor",
"1");
+ brokerConfig.setProperty("transaction.state.log.min.isr", "1");
+ brokerConfig.setProperty("num.partitions", "1");
- URL kafkaAddress = getURL(kafkaValue);
- Properties brokerConfig = properties;
+ String listener = toExternalListenerAddress(configuredBootstrap);
- for (int attemptCount = 0; attemptCount < MAX_RETRY_TO_ACQUIRE_PORT;
attemptCount++) {
- try {
- brokerConfig.setProperty("broker.id", "1");
- brokerConfig.setProperty("host.name", kafkaAddress.getHost());
- brokerConfig.setProperty("port",
String.valueOf(kafkaAddress.getPort()));
- brokerConfig.setProperty("log.dirs",
constructDir("kafka").getAbsolutePath());
- brokerConfig.setProperty("log.flush.interval.messages",
String.valueOf(1));
+ if (listener != null) {
+ brokerConfig.setProperty("listeners", listener +
",CONTROLLER://localhost:0");
+ brokerConfig.setProperty("advertised.listeners", listener);
+ }
- kafkaServer = new
KafkaServer(KafkaConfig.fromProps(brokerConfig), Time.SYSTEM,
Option.apply(this.getClass().getName()), false);
+ String replicationFactor =
properties.getProperty("offsets.topic.replication.factor");
- kafkaServer.startup();
- break;
- } catch (KafkaException | ZooKeeperClientException e) {
- LOG.warn("Attempt {}: kafka server with broker config {}
failed", attemptCount, brokerConfig);
+ if (replicationFactor != null) {
+ brokerConfig.setProperty("offsets.topic.replication.factor",
replicationFactor);
+ }
- if (attemptCount == MAX_RETRY_TO_ACQUIRE_PORT - 1) {
- throw e;
- }
+ return brokerConfig;
+ }
- if (kafkaServer != null) {
- try {
- kafkaServer.shutdown();
- } catch (Exception ex) {
- LOG.info("Failed to shutdown kafka server", ex);
- }
- }
+ private void shutdownClusterQuietly() {
+ if (cluster != null) {
+ AtomicReference<Throwable> shutdownFailure = new
AtomicReference<>();
+
+ Utils.closeQuietly(cluster, "embedded Kafka cluster",
shutdownFailure);
-
CommandHandlerUtility.tryKillingProcessUsingPort(kafkaAddress.getPort(),
attemptCount != 0);
+ if (shutdownFailure.get() != null) {
+ LOG.warn("Failed to shut down embedded Kafka cluster",
shutdownFailure.get());
}
+
+ cluster = null;
+ }
+ }
+
+ private static String toExternalListenerAddress(String bootstrapServers) {
+ if (StringUtils.isEmpty(bootstrapServers)) {
+ return null;
+ }
+
+ String hostPort = StringUtils.trim(bootstrapServers.split(",")[0]);
+
+ if (StringUtils.isEmpty(hostPort)) {
+ return null;
+ }
+
+ if (hostPort.contains("://")) {
+ return hostPort.replaceFirst("^PLAINTEXT://", "EXTERNAL://");
+ }
+
+ return "EXTERNAL://" + hostPort;
+ }
+
+ private static int parsePort(String bootstrapServers) {
+ if (StringUtils.isEmpty(bootstrapServers)) {
+ return -1;
}
- LOG.info("Embedded kafka server started with broker config {}",
brokerConfig);
+ String hostPort = StringUtils.trim(bootstrapServers.split(",")[0]);
+ int colon = hostPort.lastIndexOf(':');
+
+ if (colon < 0) {
+ return -1;
+ }
+
+ try {
+ return Integer.parseInt(hostPort.substring(colon + 1));
+ } catch (NumberFormatException e) {
+ return -1;
+ }
+ }
+
+ private void overrideExitMethods() {
+ Exit.setExitProcedure((statusCode, message) ->
+ LOG.warn("Kafka Exit.exit({}, {}) suppressed in embedded
broker", statusCode, message));
+ Exit.setHaltProcedure((statusCode, message) ->
+ LOG.warn("Kafka Exit.halt({}, {}) suppressed in embedded
broker", statusCode, message));
}
private File constructDir(String dirPrefix) {
@@ -189,12 +245,4 @@ public class EmbeddedKafkaServer implements Service {
return file;
}
-
- private URL getURL(String url) throws MalformedURLException {
- try {
- return new URL(url);
- } catch (MalformedURLException e) {
- return new URL("http://"; + url);
- }
- }
}
diff --git
a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
index 3c5ded145..e75ecfafd 100644
--- a/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
+++ b/notification/src/main/java/org/apache/atlas/kafka/KafkaNotification.java
@@ -174,9 +174,20 @@ public class KafkaNotification extends
AbstractNotification implements Service {
public void start() throws AtlasException {
LOG.info("==> KafkaNotification.start()");
+ syncBootstrapServers();
+
LOG.info("<== KafkaNotification.start()");
}
+ private void syncBootstrapServers() throws AtlasException {
+ Configuration kafkaConf =
ApplicationProperties.getSubsetConfiguration(ApplicationProperties.get(),
PROPERTY_PREFIX);
+ String bootstrap = kafkaConf.getString("bootstrap.servers");
+
+ if (StringUtils.isNotEmpty(bootstrap)) {
+ properties.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrap);
+ }
+ }
+
// ----- Service ---------------------------------------------------------
@Override
diff --git a/pom.xml b/pom.xml
index c0f359387..80eab09f3 100644
--- a/pom.xml
+++ b/pom.xml
@@ -130,8 +130,8 @@
<httpcomponents-httpclient.version>4.5.13</httpcomponents-httpclient.version>
<httpcomponents-httpcore.version>4.4.13</httpcomponents-httpcore.version>
<ivy.version>2.5.2</ivy.version>
- <jackson.databind.version>2.12.7</jackson.databind.version>
- <jackson.version>2.12.7</jackson.version>
+ <jackson.databind.version>2.16.2</jackson.databind.version>
+ <jackson.version>2.16.2</jackson.version>
<jacoco.version>0.8.13</jacoco.version>
<janusgraph.cassandra.version>0.5.3</janusgraph.cassandra.version>
<janusgraph.version>1.1.0</janusgraph.version>
@@ -149,8 +149,8 @@
<json.version>3.2.11</json.version>
<jsr.version>1.1</jsr.version>
<junit.version>4.13.2</junit.version>
- <kafka.scala.binary.version>2.12</kafka.scala.binary.version>
- <kafka.version>2.8.2</kafka.version>
+ <kafka.scala.binary.version>2.13</kafka.scala.binary.version>
+ <kafka.version>3.9.1</kafka.version>
<keycloak.version>6.0.1</keycloak.version>
<logback.version>1.3.15</logback.version>
<lucene-solr.version>8.11.3</lucene-solr.version>
@@ -1351,6 +1351,13 @@
<forkCount>0</forkCount>
<redirectTestOutputToFile>true</redirectTestOutputToFile>
</configuration>
+ <dependencies>
+ <dependency>
+ <groupId>org.apache.maven.surefire</groupId>
+ <artifactId>surefire-testng</artifactId>
+ <version>${failsafe.version}</version>
+ </dependency>
+ </dependencies>
<executions>
<execution>
<id>integration-test</id>
diff --git a/webapp/pom.xml b/webapp/pom.xml
index f28f92c45..968fef620 100755
--- a/webapp/pom.xml
+++ b/webapp/pom.xml
@@ -58,6 +58,19 @@
<version>${jackson.version}</version>
</dependency>
+ <!-- Jackson JAX-RS 2.12.7: last line compatible with Jersey 1.19 /
JAX-RS 1.1; core Jackson is 2.16.2 via ${jackson.version}. Full unification ->
ATLAS-5190 -->
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-base</artifactId>
+ <version>2.12.7</version>
+ </dependency>
+
+ <dependency>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-json-provider</artifactId>
+ <version>2.12.7</version>
+ </dependency>
+
<dependency>
<groupId>com.googlecode.json-simple</groupId>
<artifactId>json-simple</artifactId>
@@ -124,6 +137,12 @@
<artifactId>javax.servlet-api</artifactId>
</dependency>
+ <dependency>
+ <groupId>javax.ws.rs</groupId>
+ <artifactId>jsr311-api</artifactId>
+ <version>1.1.1</version>
+ </dependency>
+
<dependency>
<groupId>joda-time</groupId>
<artifactId>joda-time</artifactId>
@@ -190,6 +209,16 @@
<dependency>
<groupId>org.apache.atlas</groupId>
<artifactId>atlas-intg</artifactId>
+ <exclusions>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-base</artifactId>
+ </exclusion>
+ <exclusion>
+ <groupId>com.fasterxml.jackson.jaxrs</groupId>
+ <artifactId>jackson-jaxrs-json-provider</artifactId>
+ </exclusion>
+ </exclusions>
</dependency>
<dependency>
@@ -370,7 +399,7 @@
</exclusion>
</exclusions>
</dependency>
- <!-- PAM -->
+
<dependency>
<groupId>org.kohsuke</groupId>
<artifactId>libpam4j</artifactId>
@@ -490,7 +519,7 @@
<version>3.4</version>
<scope>test</scope>
</dependency>
-
+ <!-- PAM -->
<dependency>
<groupId>org.apache.atlas</groupId>
<artifactId>atlas-testtools</artifactId>
diff --git
a/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
b/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
index 9fb19960a..c0fdfddf5 100644
---
a/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
+++
b/webapp/src/test/java/org/apache/atlas/notification/ImportTaskListenerImplTest.java
@@ -117,7 +117,8 @@ public class ImportTaskListenerImplTest {
}
@AfterMethod
- public void teardown() {
+ public void teardown() throws Exception {
+ shutdownImportExecutor(importTaskListener);
Mockito.reset(asyncImportService, notificationHookConsumer,
requestQueue, importRequest);
}
@@ -295,6 +296,14 @@ public class ImportTaskListenerImplTest {
Mockito.doReturn("import123").when(importRequest).getImportId();
when(importRequest.getStatus()).thenReturn(WAITING);
when(importRequest.getTopicName()).thenReturn("topic1");
+
when(asyncImportService.fetchImportRequestByImportId("import123")).thenReturn(importRequest);
+
+ CountDownLatch consumerStarted = new CountDownLatch(1);
+
+ doAnswer(invocation -> {
+ consumerStarted.countDown();
+ return null;
+ }).when(notificationHookConsumer).startAsyncImportConsumer(any(),
anyString(), anyString());
ExecutorService realExecutor =
java.util.concurrent.Executors.newSingleThreadExecutor();
Field executorField =
ImportTaskListenerImpl.class.getDeclaredField("executorService");
@@ -305,22 +314,21 @@ public class ImportTaskListenerImplTest {
importTaskListener.onReceiveImportRequest(importRequest);
- Thread.sleep(500);
-
- verify(notificationHookConsumer, atLeastOnce())
-
.startAsyncImportConsumer(NotificationInterface.NotificationType.ASYNC_IMPORT,
"import123", "topic1");
+ assertTrue(consumerStarted.await(5, TimeUnit.SECONDS),
"startAsyncImportConsumer was not invoked");
- realExecutor.shutdownNow();
+ verify(notificationHookConsumer,
times(1)).startAsyncImportConsumer(NotificationInterface.NotificationType.ASYNC_IMPORT,
"import123", "topic1");
}
@Test
public void testStartImportConsumer_Failure() throws Exception {
+ when(importRequest.getImportId()).thenReturn("import123");
when(importRequest.getStatus()).thenReturn(WAITING);
when(importRequest.getTopicName()).thenReturn("topic1");
+
when(asyncImportService.fetchImportRequestByImportId("import123")).thenReturn(importRequest);
+
+ CountDownLatch consumerClosed = new CountDownLatch(1);
- doThrow(new RuntimeException("Consumer failed"))
- .when(notificationHookConsumer)
-
.startAsyncImportConsumer(NotificationInterface.NotificationType.ASYNC_IMPORT,
"import123", "topic1");
+ doThrow(new RuntimeException("Consumer
failed")).when(notificationHookConsumer).startAsyncImportConsumer(NotificationInterface.NotificationType.ASYNC_IMPORT,
"import123", "topic1");
doAnswer(invocation -> {
Object newStatus = invocation.getArgument(0);
@@ -328,6 +336,11 @@ public class ImportTaskListenerImplTest {
return null;
}).when(importRequest).setStatus(any());
+ doAnswer(invocation -> {
+ consumerClosed.countDown();
+ return null;
+ }).when(notificationHookConsumer).closeImportConsumer(anyString(),
anyString());
+
ExecutorService realExecutor =
java.util.concurrent.Executors.newSingleThreadExecutor();
Field executorField =
ImportTaskListenerImpl.class.getDeclaredField("executorService");
@@ -338,12 +351,9 @@ public class ImportTaskListenerImplTest {
importTaskListener.onReceiveImportRequest(importRequest);
- Thread.sleep(500);
-
- verify(notificationHookConsumer, times(1))
- .closeImportConsumer("import123", "ATLAS_IMPORT_import123");
+ assertTrue(consumerClosed.await(5, TimeUnit.SECONDS),
"closeImportConsumer was not invoked");
- realExecutor.shutdownNow();
+ verify(notificationHookConsumer,
times(1)).closeImportConsumer("import123", "ATLAS_IMPORT_import123");
}
@Test(dataProvider = "importQueueScenarios")
@@ -734,6 +744,20 @@ public class ImportTaskListenerImplTest {
callers.shutdownNow();
}
+ private void shutdownImportExecutor(ImportTaskListenerImpl listener)
throws Exception {
+ if (listener == null) {
+ return;
+ }
+ Field executorField =
ImportTaskListenerImpl.class.getDeclaredField("executorService");
+ executorField.setAccessible(true);
+ ExecutorService exec = (ExecutorService) executorField.get(listener);
+ if (exec != null) {
+ exec.shutdownNow();
+ exec.awaitTermination(5, TimeUnit.SECONDS);
+ executorField.set(listener, null);
+ }
+ }
+
private void setExecutorServiceAndSemaphore(ImportTaskListenerImpl
importTaskListener, ExecutorService mockExecutor, Semaphore mockSemaphore) {
try {
Field executorField =
ImportTaskListenerImpl.class.getDeclaredField("executorService");
diff --git
a/webapp/src/test/java/org/apache/atlas/web/integration/BasicSearchIT.java
b/webapp/src/test/java/org/apache/atlas/web/integration/BasicSearchIT.java
index 24c7b1924..e89eeca82 100644
--- a/webapp/src/test/java/org/apache/atlas/web/integration/BasicSearchIT.java
+++ b/webapp/src/test/java/org/apache/atlas/web/integration/BasicSearchIT.java
@@ -57,7 +57,7 @@ import static org.testng.Assert.assertTrue;
import static org.testng.Assert.fail;
public class BasicSearchIT extends BaseResourceIT {
- private static final String IMPORTED_DATA_QUALIFIED_NAME_MARKER = "@cl1";
+ private static final String IMPORTED_DATA_QUALIFIED_NAME_MARKER =
"default.";
private AtlasUserSavedSearch userSavedSearch;
@@ -350,6 +350,10 @@ public class BasicSearchIT extends BaseResourceIT {
return parameters;
}
+ if (StringUtils.isNotBlank(parameters.getQuery()) ||
parameters.getOffset() > 0) {
+ return parameters;
+ }
+
SearchParameters.FilterCriteria importedDataFilter = new
SearchParameters.FilterCriteria();
importedDataFilter.setAttributeName("qualifiedName");
diff --git
a/webapp/src/test/java/org/apache/atlas/web/resources/AdminExportImportTestIT.java
b/webapp/src/test/java/org/apache/atlas/web/resources/AdminExportImportTestIT.java
index e2dd93919..f91947f68 100644
---
a/webapp/src/test/java/org/apache/atlas/web/resources/AdminExportImportTestIT.java
+++
b/webapp/src/test/java/org/apache/atlas/web/resources/AdminExportImportTestIT.java
@@ -105,7 +105,17 @@ public class AdminExportImportTestIT extends
BaseResourceIT {
try {
performImport(FILE_TO_IMPORT, request, 32); // initial import has
5 entities already in deleted state, hence current import will have 32
processed-entities
} catch (AtlasServiceException e) {
+ if (isConnectionClosedDuringTeardown(e)) {
+ LOG.warn("performTeardown: server connection closed during
cleanup import, skipping.");
+ return;
+ }
throw new SkipException("performTeardown: failed! Subsequent tests
results may be affected.");
+ } catch (Exception e) {
+ if (isConnectionClosedDuringTeardown(e)) {
+ LOG.warn("performTeardown: server connection closed during
cleanup import, skipping.");
+ return;
+ }
+ throw e;
}
}
@@ -146,4 +156,16 @@ public class AdminExportImportTestIT extends
BaseResourceIT {
assertNotNull(server.getAdditionalInfo());
assertTrue(server.getAdditionalInfo().size() > 0);
}
+
+ private boolean isConnectionClosedDuringTeardown(Throwable error) {
+ for (Throwable current = error; current != null; current =
current.getCause()) {
+ if (current instanceof IOException && "Stream
Closed".equals(current.getMessage())) {
+ return true;
+ }
+ if (current.getMessage() != null &&
current.getMessage().contains("Stream Closed")) {
+ return true;
+ }
+ }
+ return false;
+ }
}
