This is an automated email from the ASF dual-hosted git repository.
mandarambawane 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 e61a24f75 ATLAS-5377: Remove duplicate NotificationREST endpoint from
webapp (#730)
e61a24f75 is described below
commit e61a24f75eea9cd7f320b5e415d90f21f22bafbb
Author: UmeshPatil-1 <[email protected]>
AuthorDate: Wed Sep 16 13:24:52 2026 +0530
ATLAS-5377: Remove duplicate NotificationREST endpoint from webapp (#730)
* ATLAS-5377: Remove duplicate NotificationREST endpoint from the main
webapp (port 21000) and require hooks to use rest-notification-webapp (port
41000/rest) only.
* ATLAS-5377: Updated NotificationRestIT to assert endpoint is removed from
webapp (expect error, not 204).
* ATLAS-5377: Updated assertion and shorter fail-fast message as per review.
---
.../main/java/org/apache/atlas/AtlasConstants.java | 3 +-
.../atlas/notification/rest/RestNotification.java | 10 +-
.../atlas/notification/RestNotificationTest.java | 44 ++++---
.../apache/atlas/web/rest/NotificationREST.java | 132 ---------------------
.../atlas/web/integration/NotificationRestIT.java | 32 +++--
5 files changed, 53 insertions(+), 168 deletions(-)
diff --git a/common/src/main/java/org/apache/atlas/AtlasConstants.java
b/common/src/main/java/org/apache/atlas/AtlasConstants.java
index cc0d50763..e2930062c 100644
--- a/common/src/main/java/org/apache/atlas/AtlasConstants.java
+++ b/common/src/main/java/org/apache/atlas/AtlasConstants.java
@@ -31,7 +31,8 @@ public final class AtlasConstants {
public static final String ATLAS_SERVICES_ENABLED =
"atlas.services.enabled";
public static final String CLUSTER_NAME_ATTRIBUTE = "clusterName";
public static final String DEFAULT_APP_PORT_STR = "21000";
- public static final String DEFAULT_ATLAS_REST_ADDRESS =
"http://localhost:21000";;
+ public static final String DEFAULT_ATLAS_REST_ADDRESS =
"http://localhost:21000";;
+ public static final String DEFAULT_REST_NOTIFICATION_ADDRESS =
"http://localhost:41000/rest";;
public static final String DEFAULT_TYPE_VERSION = "1.0";
public static final int ATLAS_SHUTDOWN_HOOK_PRIORITY = 30;
diff --git
a/notification/src/main/java/org/apache/atlas/notification/rest/RestNotification.java
b/notification/src/main/java/org/apache/atlas/notification/rest/RestNotification.java
index e94afddc9..0743545ff 100644
---
a/notification/src/main/java/org/apache/atlas/notification/rest/RestNotification.java
+++
b/notification/src/main/java/org/apache/atlas/notification/rest/RestNotification.java
@@ -47,10 +47,9 @@ public class RestNotification extends AbstractNotification {
private static final Logger LOG =
LoggerFactory.getLogger(RestNotification.class);
private static final int BATCH_MAX_LENGTH_BYTES =
AtlasConfiguration.NOTIFICATION_REST_BODY_MAX_LENGTH_BYTES.getInt();
- private static final String ATLAS_HOOK_REST_NOTIFICATION_ENDPOINT =
"atlas.hook.rest.notification.address";
+ private static final String ATLAS_HOOK_REST_NOTIFICATION_ENDPOINT =
AtlasConfiguration.NOTIFICATION_HOOK_REST_ADDRESS.getPropertyName();
private static final String BASIC_AUTH_USERNAME =
"atlas.rest.basic.auth.username";
private static final String BASIC_AUTH_PASSWORD =
"atlas.rest.basic.auth.password";
- private static final String DEFAULT_ATLAS_URL =
"http://localhost:31000/";;
private static final Map<NotificationType, String> PRODUCER_TOPIC_MAP =
new HashMap<>();
@@ -115,11 +114,8 @@ public class RestNotification extends AbstractNotification
{
String[] atlasEndPoint =
configuration.getStringArray(ATLAS_HOOK_REST_NOTIFICATION_ENDPOINT);
if (isEndpointNotSpecified(atlasEndPoint)) {
- atlasEndPoint =
configuration.getStringArray(AtlasConstants.ATLAS_REST_ADDRESS_KEY);
- }
-
- if (isEndpointNotSpecified(atlasEndPoint)) {
- atlasEndPoint = new String[] {DEFAULT_ATLAS_URL};
+ throw new AtlasException(ATLAS_HOOK_REST_NOTIFICATION_ENDPOINT
+ " is required for rest-notification-webapp (e.g. "
+ + AtlasConstants.DEFAULT_REST_NOTIFICATION_ADDRESS +
")");
}
if (!AuthenticationUtil.isKerberosAuthenticationEnabled()) {
diff --git
a/notification/src/test/java/org/apache/atlas/notification/RestNotificationTest.java
b/notification/src/test/java/org/apache/atlas/notification/RestNotificationTest.java
index 417adc7df..5db77523b 100644
---
a/notification/src/test/java/org/apache/atlas/notification/RestNotificationTest.java
+++
b/notification/src/test/java/org/apache/atlas/notification/RestNotificationTest.java
@@ -25,6 +25,7 @@ import org.apache.atlas.AtlasBaseClient;
import org.apache.atlas.AtlasClientV2;
import org.apache.atlas.AtlasConfiguration;
import org.apache.atlas.AtlasErrorCode;
+import org.apache.atlas.AtlasException;
import org.apache.atlas.kafka.NotificationProvider;
import org.apache.atlas.notification.rest.RestNotification;
import org.apache.commons.configuration2.BaseConfiguration;
@@ -68,6 +69,7 @@ public class RestNotificationTest {
conf = ApplicationProperties.get();
conf.setProperty(AtlasConfiguration.NOTIFICATION_HOOK_REST_ENABLED.getPropertyName(),
true);
+
conf.setProperty(AtlasConfiguration.NOTIFICATION_HOOK_REST_ADDRESS.getPropertyName(),
"http://localhost:41000/rest";);
conf.setProperty(NotificationProvider.CONF_ATLAS_HOOK_SPOOL_ENABLED,
false);
notifier = NotificationProvider.get();
@@ -150,23 +152,17 @@ public class RestNotificationTest {
assertEquals(configuredEndpoints[0],
"http://atlas-rest.example.com:41000/rest";);
}
- @Test
- public void testRestNotificationEndpointFallsBackToAtlasRestAddress()
throws Exception {
+ @Test(expectedExceptions = AtlasException.class)
+ public void testRestNotificationFailsFastWhenHookAddressNotConfigured()
throws Exception {
Configuration localConf = new BaseConfiguration();
- localConf.setProperty("atlas.rest.address",
"http://atlas-main.example.com:21000";);
localConf.setProperty("atlas.rest.basic.auth.username", "admin");
localConf.setProperty("atlas.rest.basic.auth.password", "admin123");
- RestNotification restNotification = new RestNotification(localConf);
-
- String[] configuredEndpoints =
getConfiguredBaseUrls(restNotification.atlasClientV2);
-
- assertEquals(configuredEndpoints.length, 1);
- assertEquals(configuredEndpoints[0],
"http://atlas-main.example.com:21000";);
+ new RestNotification(localConf);
}
- @Test
- public void
testRestNotificationFallsBackToAtlasRestWhenNotificationAddressesAllBlank()
throws Exception {
+ @Test(expectedExceptions = AtlasException.class)
+ public void testRestNotificationFailsFastWhenHookAddressesAllBlank()
throws Exception {
Configuration localConf = new BaseConfiguration();
localConf.addProperty("atlas.hook.rest.notification.address", "");
localConf.addProperty("atlas.hook.rest.notification.address", " ");
@@ -174,12 +170,30 @@ public class RestNotificationTest {
localConf.setProperty("atlas.rest.basic.auth.username", "admin");
localConf.setProperty("atlas.rest.basic.auth.password", "admin123");
- RestNotification restNotification = new RestNotification(localConf);
+ new RestNotification(localConf);
+ }
- String[] configuredEndpoints =
getConfiguredBaseUrls(restNotification.atlasClientV2);
+ @Test(expectedExceptions = AtlasException.class)
+ public void
testRestNotificationFailsFastWhenOnlyAtlasRestAddressConfigured() throws
Exception {
+ Configuration localConf = new BaseConfiguration();
+ localConf.setProperty("atlas.rest.address",
"http://atlas-main.example.com:21000";);
+ localConf.setProperty("atlas.rest.basic.auth.username", "admin");
+ localConf.setProperty("atlas.rest.basic.auth.password", "admin123");
- assertEquals(configuredEndpoints.length, 1);
- assertEquals(configuredEndpoints[0],
"http://atlas-main.example.com:21000";);
+ new RestNotification(localConf);
+ }
+
+ @Test
+ public void testRestNotificationFailFastExceptionMessage() {
+ Configuration localConf = new BaseConfiguration();
+
+ try {
+ new RestNotification(localConf);
+ fail("Expected AtlasException when
atlas.hook.rest.notification.address is not configured");
+ } catch (AtlasException e) {
+
assertTrue(e.getMessage().contains("atlas.hook.rest.notification.address is
required"));
+ assertTrue(e.getMessage().contains("http://localhost:41000/rest";));
+ }
}
private String[] getConfiguredBaseUrls(AtlasClientV2 atlasClientV2) throws
Exception {
diff --git
a/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java
b/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java
deleted file mode 100644
index 6c437ea98..000000000
--- a/webapp/src/main/java/org/apache/atlas/web/rest/NotificationREST.java
+++ /dev/null
@@ -1,132 +0,0 @@
-/**
- * Licensed to the Apache Software Foundation (ASF) under one
- * or more contributor license agreements. See the NOTICE file
- * distributed with this work for additional information
- * regarding copyright ownership. The ASF licenses this file
- * to you under the Apache License, Version 2.0 (the
- * "License"); you may not use this file except in compliance
- * with the License. You may obtain a copy of the License at
- * <p>
- * http://www.apache.org/licenses/LICENSE-2.0
- * <p>
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.apache.atlas.web.rest;
-
-import com.fasterxml.jackson.databind.JsonNode;
-import com.fasterxml.jackson.databind.node.ArrayNode;
-import org.apache.atlas.AtlasConfiguration;
-import org.apache.atlas.AtlasErrorCode;
-import org.apache.atlas.authorize.AtlasAdminAccessRequest;
-import org.apache.atlas.authorize.AtlasAuthorizationUtils;
-import org.apache.atlas.authorize.AtlasPrivilege;
-import org.apache.atlas.exception.AtlasBaseException;
-import org.apache.atlas.hook.AtlasHook;
-import org.apache.atlas.kafka.KafkaNotification;
-import org.apache.atlas.notification.NotificationException;
-import org.apache.atlas.notification.NotificationInterface;
-import org.apache.atlas.server.common.util.Servlets;
-import org.apache.atlas.utils.AtlasJson;
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-import org.springframework.stereotype.Service;
-
-import javax.inject.Inject;
-import javax.inject.Singleton;
-import javax.servlet.http.HttpServletRequest;
-import javax.ws.rs.Consumes;
-import javax.ws.rs.POST;
-import javax.ws.rs.Path;
-import javax.ws.rs.PathParam;
-import javax.ws.rs.Produces;
-import javax.ws.rs.core.Context;
-import javax.ws.rs.core.MediaType;
-
-import java.io.IOException;
-import java.util.ArrayList;
-import java.util.Arrays;
-import java.util.HashSet;
-import java.util.List;
-import java.util.Set;
-
-@Path("v2/notification")
-@Singleton
-@Service
-@Consumes({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
-@Produces({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
-public class NotificationREST {
- private static final Logger LOG =
LoggerFactory.getLogger(NotificationREST.class);
-
- public static final String ATLAS_HOOK_TOPIC =
AtlasConfiguration.NOTIFICATION_HOOK_TOPIC_NAME.getString();
- public static final String ATLAS_ENTITIES_TOPIC =
AtlasConfiguration.NOTIFICATION_ENTITIES_TOPIC_NAME.getString();
-
- private static final String[] ATLAS_HOOK_CONSUMER_TOPICS =
AtlasConfiguration.NOTIFICATION_HOOK_CONSUMER_TOPIC_NAMES.getStringArray(ATLAS_HOOK_TOPIC);
- private static final String[] ATLAS_ENTITIES_CONSUMER_TOPICS =
AtlasConfiguration.NOTIFICATION_ENTITIES_CONSUMER_TOPIC_NAMES.getStringArray(ATLAS_ENTITIES_TOPIC);
- private static final Set<String> TOPICS = new
HashSet<>();
-
- private final NotificationInterface notificationInterface;
-
- @Inject
- public NotificationREST(NotificationInterface notificationInterface) {
- this.notificationInterface = notificationInterface;
- }
-
- /**
- * Publish notifications on Kafka topic
- *
- * @param topicName - nameOfTheQueue
- * @throws AtlasBaseException
- */
- @POST
- @Path("/topic/{topicName}")
- @Consumes({Servlets.JSON_MEDIA_TYPE, MediaType.APPLICATION_JSON})
- public void handleNotifications(@PathParam("topicName") String topicName,
@Context HttpServletRequest request) throws AtlasBaseException, IOException {
- LOG.debug("Handling notifications for topic {}", topicName);
-
- AtlasAuthorizationUtils.verifyAccess(new
AtlasAdminAccessRequest(AtlasPrivilege.SERVICE_NOTIFICATION_POST), "post on
rest notification service");
-
- if (!TOPICS.contains(topicName)) {
- throw new AtlasBaseException(AtlasErrorCode.INVALID_TOPIC_NAME,
topicName);
- }
-
- String messagesAsJson = Servlets.getRequestPayload(request);
- List<String> messages = getMessagesToNotify(messagesAsJson);
-
- try {
- KafkaNotification notifier = (KafkaNotification)
notificationInterface;
-
- notifier.sendInternal(topicName, messages,
AtlasHook.isHookMsgsSortEnabled);
- } catch (NotificationException exception) {
- List<String> failedMessages = exception.getFailedMessages();
- String concatenatedMessage =
StringUtils.join(failedMessages, "\n");
-
- throw new
AtlasBaseException(AtlasErrorCode.NOTIFICATION_EXCEPTION, exception,
concatenatedMessage);
- }
- }
-
- private List<String> getMessagesToNotify(String messagesAsJson) {
- List<String> messages = new ArrayList<>();
-
- try {
- ArrayNode messageNodes =
AtlasJson.parseToV1ArrayNode(messagesAsJson);
-
- for (JsonNode messageNode : messageNodes) {
- messages.add(AtlasJson.toV1Json(messageNode));
- }
- } catch (IOException e) {
- messages.add(messagesAsJson);
- }
-
- return messages;
- }
-
- static {
- TOPICS.addAll(Arrays.asList(ATLAS_HOOK_CONSUMER_TOPICS));
- TOPICS.addAll(Arrays.asList(ATLAS_ENTITIES_CONSUMER_TOPICS));
- }
-}
diff --git
a/webapp/src/test/java/org/apache/atlas/web/integration/NotificationRestIT.java
b/webapp/src/test/java/org/apache/atlas/web/integration/NotificationRestIT.java
index 96df5d599..54c1a4bce 100644
---
a/webapp/src/test/java/org/apache/atlas/web/integration/NotificationRestIT.java
+++
b/webapp/src/test/java/org/apache/atlas/web/integration/NotificationRestIT.java
@@ -17,7 +17,7 @@
*/
package org.apache.atlas.web.integration;
-import com.fasterxml.jackson.databind.node.ArrayNode;
+import com.sun.jersey.api.client.ClientResponse;
import org.apache.atlas.AtlasClientV2;
import org.apache.atlas.AtlasServiceException;
import org.apache.atlas.utils.TestResourceFileUtils;
@@ -30,40 +30,46 @@ import java.util.Date;
import static org.apache.atlas.kafka.KafkaNotification.ATLAS_HOOK_TOPIC;
import static org.testng.Assert.assertNotNull;
-import static org.testng.Assert.assertNull;
+import static org.testng.Assert.assertTrue;
+import static org.testng.Assert.fail;
+/**
+ * ATLAS-5377: hook notification REST ingress was removed from the main Atlas
webapp.
+ * Canonical endpoint is rest-notification-webapp only. These integration
tests verify
+ * that POST /api/atlas/v2/notification/topic/{topicName} is not served on
main webapp.
+ */
public class NotificationRestIT extends BaseResourceIT {
@Test
- public void unAuthPostNotification() throws IOException {
+ public void unAuthPostNotificationRejected() throws IOException {
AtlasClientV2 unAuthClient = new AtlasClientV2(atlasUrls, new String[]
{"admin", "wr0ng_pa55w0rd"});
try {
- unAuthClient.postNotificationToTopic(ATLAS_HOOK_TOPIC, new
ArrayList<String>(Collections.singletonList("Dummy")));
+ unAuthClient.postNotificationToTopic(ATLAS_HOOK_TOPIC, new
ArrayList<>(Collections.singletonList("Dummy")));
+
+ fail("Expected postNotificationToTopic to fail on main webapp");
} catch (AtlasServiceException e) {
assertNotNull(e.getStatus(), "expected server error code in the
status");
}
}
@Test
- public void postNotificationBasicTest() throws Exception {
+ public void notificationEndpointRemovedFromMainWebapp() throws Exception {
String dbName = "db_" + randomString();
String clusterName = "cl" + randomString();
- String qualifiedName = dbName + "@" + clusterName;
String notificationString =
TestResourceFileUtils.getJson("notifications/create-db")
.replaceAll("--name--", dbName).replaceAll("--clName--",
clusterName)
.replace("\"--ts--\"", String.valueOf((new Date()).getTime()));
try {
- atlasClientV2.postNotificationToTopic(ATLAS_HOOK_TOPIC, new
ArrayList<String>(Collections.singletonList(notificationString)));
+ atlasClientV2.postNotificationToTopic(ATLAS_HOOK_TOPIC, new
ArrayList<>(Collections.singletonList(notificationString)));
- waitFor(MAX_WAIT_TIME, () -> {
- ArrayNode results = searchByDSL(String.format("%s where
qualifiedName='%s'", DATABASE_TYPE_BUILTIN, qualifiedName));
-
- return results.size() == 1;
- });
+ fail("Expected notification POST to fail — endpoint removed from
main webapp (ATLAS-5377)");
} catch (AtlasServiceException e) {
- assertNull(e.getStatus(), "expected no server error code in the
status");
+ assertNotNull(e.getStatus(), "expected HTTP error when posting to
removed endpoint");
+
+ assertTrue(e.getStatus().getStatusCode() !=
ClientResponse.Status.NO_CONTENT.getStatusCode(),
+ "notification POST must not return 204 on main webapp");
}
}
}
