diff --git a/addons/sqoop-bridge/pom.xml b/addons/sqoop-bridge/pom.xml index ea1bc94bab3..f4954624cc2 100644 --- a/addons/sqoop-bridge/pom.xml +++ b/addons/sqoop-bridge/pom.xml @@ -34,6 +34,12 @@ + + com.fasterxml.jackson.core + jackson-databind + ${jackson.databind.version} + + org.apache.atlas atlas-client-v1 @@ -129,6 +135,13 @@ + + com.fasterxml.jackson.core + jackson-core + ${jackson.version} + test + + org.apache.atlas atlas-graphdb-impls diff --git a/addons/storm-bridge/pom.xml b/addons/storm-bridge/pom.xml index 42e325f1c72..1033a373960 100644 --- a/addons/storm-bridge/pom.xml +++ b/addons/storm-bridge/pom.xml @@ -33,6 +33,12 @@ Apache Atlas Storm Bridge Module + + + com.fasterxml.jackson.core + jackson-databind + ${jackson.databind.version} + commons-collections @@ -55,7 +61,7 @@ metrics-graphite ${dropwizard-metrics} - + org.apache.atlas atlas-client-v1 diff --git a/addons/trino-extractor/pom.xml b/addons/trino-extractor/pom.xml index b66b957ca54..9a15f3edf08 100644 --- a/addons/trino-extractor/pom.xml +++ b/addons/trino-extractor/pom.xml @@ -35,6 +35,8 @@ true false + + true 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 daa600949a9..c7260ff6c5c 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 void addClassification(ClassificationAssociateRequest request) throws Atl } public void addClassifications(String guid, List 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 uniqAttributes, List classifications) throws AtlasServiceException { MultivaluedMap 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 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 uniqAttributes, List classifications) throws AtlasServiceException { MultivaluedMap 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 c07b9fb6368..2041f68aa3b 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 5c5dba8020c..4406fbd90d0 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 5c22fe73b39..17ea80aff6e 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 126143a204d..35ba5a3db4f 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 1885f7c8d61..22c1f6021fa 100644 --- a/notification/pom.xml +++ b/notification/pom.xml @@ -63,21 +63,49 @@ org.apache.kafka kafka-clients - + + org.apache.kafka + kafka-clients + ${kafka.version} + test + + + org.apache.kafka + kafka-metadata + ${kafka.version} + + + org.apache.kafka + kafka-server + ${kafka.version} + + + org.apache.kafka + kafka-server-common + ${kafka.version} + + + org.apache.kafka + kafka-server-common + ${kafka.version} + test + org.apache.kafka kafka_${kafka.scala.binary.version} - - org.mockito - mockito-core - ${mockito.version} + org.apache.kafka + kafka_${kafka.scala.binary.version} + ${kafka.version} + test + - org.testng - testng + org.junit.jupiter + junit-jupiter-api + 5.11.4 @@ -94,6 +122,19 @@ test + + org.mockito + mockito-core + ${mockito.version} + test + + + + org.testng + testng + test + + @@ -172,7 +213,7 @@ org.apache.zookeeper zookeeper - 3.4.6 + ${zookeeper.version} 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 23158f521b7..654caa7d9ff 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 void start() throws AtlasException { if (isEmbedded) { try { - startZk(); - startKafka(); + startKraftBroker(); } catch (Exception e) { throw new AtlasException("Failed to start embedded kafka", e); } @@ -94,90 +88,152 @@ public void start() throws AtlasException { 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 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 @@ 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 index 3c5ded14535..e75ecfafd65 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 static List 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 index c0f35938789..80eab09f3f2 100644 --- a/pom.xml +++ b/pom.xml @@ -130,8 +130,8 @@ 4.5.13 4.4.13 2.5.2 - 2.12.7 - 2.12.7 + 2.16.2 + 2.16.2 0.8.13 0.5.3 1.1.0 @@ -149,8 +149,8 @@ 3.2.11 1.1 4.13.2 - 2.12 - 2.8.2 + 2.13 + 3.9.1 6.0.1 1.3.15 8.11.3 @@ -1351,6 +1351,13 @@ 0 true + + + org.apache.maven.surefire + surefire-testng + ${failsafe.version} + + integration-test diff --git a/webapp/pom.xml b/webapp/pom.xml index f28f92c4561..968fef62019 100755 --- a/webapp/pom.xml +++ b/webapp/pom.xml @@ -58,6 +58,19 @@ ${jackson.version} + + + com.fasterxml.jackson.jaxrs + jackson-jaxrs-base + 2.12.7 + + + + com.fasterxml.jackson.jaxrs + jackson-jaxrs-json-provider + 2.12.7 + + com.googlecode.json-simple json-simple @@ -124,6 +137,12 @@ javax.servlet-api + + javax.ws.rs + jsr311-api + 1.1.1 + + joda-time joda-time @@ -190,6 +209,16 @@ org.apache.atlas atlas-intg + + + com.fasterxml.jackson.jaxrs + jackson-jaxrs-base + + + com.fasterxml.jackson.jaxrs + jackson-jaxrs-json-provider + + @@ -370,7 +399,7 @@ - + org.kohsuke libpam4j @@ -490,7 +519,7 @@ 3.4 test - + org.apache.atlas atlas-testtools 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 9fb19960ab5..c0fdfddf574 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 void resetMocks() throws AtlasException { } @AfterMethod - public void teardown() { + public void teardown() throws Exception { + shutdownImportExecutor(importTaskListener); Mockito.reset(asyncImportService, notificationHookConsumer, requestQueue, importRequest); } @@ -295,6 +296,14 @@ public void testStartImportConsumer_Successful() throws Exception { 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 void testStartImportConsumer_Successful() throws Exception { 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 void testStartImportConsumer_Failure() throws Exception { 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 void testStartImportConsumer_Failure() throws Exception { 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 void ensureExecutorAliveReturnsNullWhenPassiveEvenUnderConcurrency() thro 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 24c7b192489..e89eeca8293 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.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 @@ private SearchParameters scopeToImportedDataset(SearchParameters parameters) { 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 e2dd93919f2..f91947f684a 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 void teardown() { 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 @@ private void assertReplicationData(String serverName) throws AtlasServiceExcepti 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; + } }