apache · sarathsubramanian · Jul 8, 2026 · Jul 4, 2026 · Jul 6, 2026 · Jul 6, 2026
diff --git 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
@@ -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
@@ -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
@@ -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
@@ -63,10 +63,48 @@
             <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.apache.kafka</groupId>
             <artifactId>kafka_${kafka.scala.binary.version}</artifactId>
+            <version>${kafka.version}</version>
+            <classifier>test</classifier>
+        </dependency>
+
+        <dependency>
+            <groupId>org.junit.jupiter</groupId>
+            <artifactId>junit-jupiter-api</artifactId>
+            <version>5.11.4</version>
         </dependency>
 
         <dependency>
@@ -172,7 +210,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
@@ -17,36 +17,27 @@
  */
 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.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)
@@ -56,17 +47,18 @@ public class EmbeddedKafkaServer implements Service {
     public static final String PROPERTY_PREFIX   = "atlas.kafka";
     public static final String PROPERTY_EMBEDDED = "atlas.notification.embedded";
 
-    private static final String ATLAS_KAFKA_DATA          = "data";
-    private static final int    MAX_RETRY_TO_ACQUIRE_PORT = 3;
+    private static final String ATLAS_KAFKA_DATA = "data";
 
-    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 +70,7 @@ public void start() throws AtlasException {
 
         if (isEmbedded) {
             try {
-                startZk();
-                startKafka();
+                startKraftBroker();
             } catch (Exception e) {
                 throw new AtlasException("Failed to start embedded kafka", e);
             }
@@ -94,90 +85,77 @@ public void start() throws AtlasException {
     public void stop() {
         LOG.info("==> EmbeddedKafkaServer.stop(isEmbedded={})", isEmbedded);
 
-        if (kafkaServer != null) {
-            kafkaServer.shutdown();
-        }
+        if (cluster != null) {
+            AtomicReference<Throwable> shutdownFailure = new AtomicReference<>();
+
+            Utils.closeQuietly(cluster, "embedded Kafka cluster", shutdownFailure);
+
+            if (shutdownFailure.get() != null) {
+                LOG.warn("Failed to shut down embedded Kafka cluster", shutdownFailure.get());
+            }
 
-        if (factory != null) {
-            factory.shutdown();
+            cluster = null;
         }
 
         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");
 
-        URL  zkAddress   = getURL(zkValue);
-        File snapshotDir = constructDir("zk/txn");
-        File logDir      = constructDir("zk/snap");
+        LOG.info("Starting embedded KRaft kafka (log.dir={})", logDir.getAbsolutePath());
 
-        for (int attemptCount = 0; attemptCount < MAX_RETRY_TO_ACQUIRE_PORT; attemptCount++) {
-            try {
-                factory = NIOServerCnxnFactory.createFactory(new InetSocketAddress(zkAddress.getHost(), zkAddress.getPort()), 1024);
-                break;
-            } catch (BindException e) {
-                LOG.warn("Attempt {}: Starting zookeeper at {} failed", attemptCount, zkValue);
+        KafkaClusterTestKit.Builder clusterBuilder = new KafkaClusterTestKit.Builder(
+                new TestKitNodes.Builder()
+                        .setCombined(true)
+                        .setNumBrokerNodes(1)
+                        .setNumControllerNodes(1)
+                        .build());
 
-                if (attemptCount == MAX_RETRY_TO_ACQUIRE_PORT - 1) {
-                    throw e;
-                }
+        Properties brokerConfig = buildBrokerConfig(logDir);
 
-                CommandHandlerUtility.tryKillingProcessUsingPort(zkAddress.getPort(), attemptCount != 0);
-            }
-        }
+        brokerConfig.forEach((key, value) -> clusterBuilder.setConfigProp(key.toString(), value));
 
-        factory.startup(new ZooKeeperServer(snapshotDir, logDir, 500));
+        cluster = clusterBuilder.build();
+        cluster.format();
+        cluster.startup();
+        cluster.waitForReadyBrokers();
 
-        String ret = factory.getLocalAddress().getAddress().toString();
+        String bootstrapServers = cluster.clientProperties().get(CommonClientConfigs.BOOTSTRAP_SERVERS_CONFIG).toString();
 
-        LOG.info("Embedded zookeeper for Kafka started at {}", ret);
+        LOG.info("Embedded KRaft kafka server started at {}", bootstrapServers);
 
-        return ret;
+        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) {
+        Properties brokerConfig = new Properties();
 
-        LOG.info("Starting kafka at {}", kafkaValue);
+        brokerConfig.setProperty("log.dir", logDir.getAbsolutePath());
+        brokerConfig.setProperty("delete.topic.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 replicationFactor = properties.getProperty("offsets.topic.replication.factor");
 
-        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));
-
-                kafkaServer = new KafkaServer(KafkaConfig.fromProps(brokerConfig), Time.SYSTEM, Option.apply(this.getClass().getName()), false);
-
-                kafkaServer.startup();
-                break;
-            } catch (KafkaException | ZooKeeperClientException e) {
-                LOG.warn("Attempt {}: kafka server with broker config {} failed", attemptCount, brokerConfig);
-
-                if (attemptCount == MAX_RETRY_TO_ACQUIRE_PORT - 1) {
-                    throw e;
-                }
-
-                if (kafkaServer != null) {
-                    try {
-                        kafkaServer.shutdown();
-                    } catch (Exception ex) {
-                        LOG.info("Failed to shutdown kafka server", ex);
-                    }
-                }
-
-                CommandHandlerUtility.tryKillingProcessUsingPort(kafkaAddress.getPort(), attemptCount != 0);
-            }
+        if (replicationFactor != null) {
+            brokerConfig.setProperty("offsets.topic.replication.factor", replicationFactor);
         }
 
-        LOG.info("Embedded kafka server started with broker config {}", brokerConfig);
+        return brokerConfig;
+    }
+
+    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 +167,4 @@ private File constructDir(String dirPrefix) {
 
         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
@@ -174,9 +174,20 @@ public static List<String> trimAndPurge(String[] strings) {
     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
@@ -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>