diff --git a/common/src/main/java/org/apache/atlas/AtlasConstants.java b/common/src/main/java/org/apache/atlas/AtlasConstants.java index cc0d507633a..e2930062ce7 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 e94afddc9e6..85b1aa4b626 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 PRODUCER_TOPIC_MAP = new HashMap<>(); @@ -115,11 +114,10 @@ private AtlasClientV2 setupAtlasClientV2(Configuration configuration) throws Atl 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.address must be configured when REST notification is enabled. " + + "Hook REST ingress is served only by rest-notification-webapp (port 41000); " + + "the main Atlas webapp notification endpoint has been removed. " + + "Example: " + 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 417adc7df8a..0bf90efc8bd 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.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 void setup() throws Exception { 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 void testRestNotificationEndpointPrefersHookRestAddress() throws Exceptio 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 void testRestNotificationFallsBackToAtlasRestWhenNotificationAddressesAll 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 must be configured")); + 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 6c437ea98b1..00000000000 --- 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 - *

- * http://www.apache.org/licenses/LICENSE-2.0 - *

- * 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 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 messages = getMessagesToNotify(messagesAsJson); - - try { - KafkaNotification notifier = (KafkaNotification) notificationInterface; - - notifier.sendInternal(topicName, messages, AtlasHook.isHookMsgsSortEnabled); - } catch (NotificationException exception) { - List failedMessages = exception.getFailedMessages(); - String concatenatedMessage = StringUtils.join(failedMessages, "\n"); - - throw new AtlasBaseException(AtlasErrorCode.NOTIFICATION_EXCEPTION, exception, concatenatedMessage); - } - } - - private List getMessagesToNotify(String messagesAsJson) { - List 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 96df5d59966..54c1a4bcead 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 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(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(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"); } } }