Remove topic blacklist feature

hermes-api/src/main/java/pl/allegro/tech/hermes/api/ErrorCode.java

@@ -48,9 +48,7 @@ public enum ErrorCode {
   SUBSCRIPTION_ENDPOINT_ADDRESS_CHANGE_EXCEPTION(INTERNAL_SERVER_ERROR),
   OAUTH_PROVIDER_NOT_EXISTS(NOT_FOUND),
   OAUTH_PROVIDER_ALREADY_EXISTS(BAD_REQUEST),
-  TOPIC_BLACKLISTED(FORBIDDEN),
   THROUGHPUT_QUOTA_VIOLATION(429),
-  TOPIC_NOT_UNBLACKLISTED(BAD_REQUEST),
   TOPIC_CONSTRAINTS_ALREADY_EXIST(BAD_REQUEST),
   TOPIC_CONSTRAINTS_DO_NOT_EXIST(BAD_REQUEST),
   SUBSCRIPTION_CONSTRAINTS_ALREADY_EXIST(BAD_REQUEST),

hermes-common/src/main/java/pl/allegro/tech/hermes/infrastructure/zookeeper/ZookeeperPaths.java

@@ -23,7 +23,6 @@ public class ZookeeperPaths {
   public static final String ADMIN_PATH = "admin";
   public static final String PREVIEW_PATH = "preview";
   public static final String OAUTH_PROVIDERS_PATH = "oauth-providers";
-  public static final String BLACKLIST_PATH = "blacklist";
   public static final String MAX_RATE_PATH = "max-rate";
   public static final String MAX_RATE_HISTORY_PATH = "history";
   public static final String STORAGE_HEALTH_PATH = "storage-health";
@@ -152,14 +151,6 @@ public String consumersWorkloadConstraintsPath(String constraintsPath) {
     return Joiner.on(URL_SEPARATOR).join(consumersWorkloadConstraintsPath(), constraintsPath);
   }
 
-  public String topicsBlacklistPath() {
-    return Joiner.on(URL_SEPARATOR).join(basePath, BLACKLIST_PATH, TOPICS_PATH);
-  }
-
-  public String blacklistedTopicPath(String qualifiedTopicName) {
-    return Joiner.on(URL_SEPARATOR).join(topicsBlacklistPath(), qualifiedTopicName);
-  }
-
   public String oAuthProvidersPath() {
     return Joiner.on(URL_SEPARATOR).join(basePath, OAUTH_PROVIDERS_PATH);
   }

hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/cache/topic/NotificationBasedTopicsCache.java

@@ -8,20 +8,16 @@
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import pl.allegro.tech.hermes.api.Topic;
-import pl.allegro.tech.hermes.api.TopicName;
 import pl.allegro.tech.hermes.common.kafka.KafkaNamesMapper;
 import pl.allegro.tech.hermes.common.metric.MetricsFacade;
 import pl.allegro.tech.hermes.domain.group.GroupRepository;
 import pl.allegro.tech.hermes.domain.notifications.InternalNotificationsBus;
 import pl.allegro.tech.hermes.domain.notifications.TopicCallback;
 import pl.allegro.tech.hermes.domain.topic.TopicRepository;
-import pl.allegro.tech.hermes.frontend.blacklist.BlacklistZookeeperNotifyingCache;
-import pl.allegro.tech.hermes.frontend.blacklist.TopicBlacklistCallback;
 import pl.allegro.tech.hermes.frontend.metric.CachedTopic;
 import pl.allegro.tech.hermes.frontend.metric.ThroughputRegistry;
 
-public class NotificationBasedTopicsCache
-    implements TopicCallback, TopicsCache, TopicBlacklistCallback {
+public class NotificationBasedTopicsCache implements TopicCallback, TopicsCache {
 
   private static final Logger logger = LoggerFactory.getLogger(NotificationBasedTopicsCache.class);
 
@@ -35,7 +31,6 @@ public class NotificationBasedTopicsCache
 
   public NotificationBasedTopicsCache(
       InternalNotificationsBus notificationsBus,
-      BlacklistZookeeperNotifyingCache blacklistZookeeperNotifyingCache,
       GroupRepository groupRepository,
       TopicRepository topicRepository,
       MetricsFacade metricsFacade,
@@ -47,7 +42,6 @@ public NotificationBasedTopicsCache(
     this.kafkaNamesMapper = kafkaNamesMapper;
     this.throughputRegistry = throughputRegistry;
     notificationsBus.registerTopicCallback(this);
-    blacklistZookeeperNotifyingCache.addCallback(this);
   }
 
   @Override
@@ -76,34 +70,6 @@ public void onTopicChanged(Topic topic) {
     topicCache.put(topic.getName().qualifiedName(), cachedTopic(topic));
   }
 
-  @Override
-  public void onTopicBlacklisted(String qualifiedTopicName) {
-    Optional<Topic> topic =
-        Optional.ofNullable(
-            Optional.ofNullable(topicCache.get(qualifiedTopicName))
-                .map(CachedTopic::getTopic)
-                .orElseGet(
-                    () ->
-                        topicRepository.getTopicDetails(
-                            TopicName.fromQualifiedName(qualifiedTopicName))));
-
-    topic.ifPresent(t -> topicCache.put(qualifiedTopicName, bannedTopic(t)));
-  }
-
-  @Override
-  public void onTopicUnblacklisted(String qualifiedTopicName) {
-    Optional<Topic> topic =
-        Optional.ofNullable(
-            Optional.ofNullable(topicCache.get(qualifiedTopicName))
-                .map(CachedTopic::getTopic)
-                .orElseGet(
-                    () ->
-                        topicRepository.getTopicDetails(
-                            TopicName.fromQualifiedName(qualifiedTopicName))));
-
-    topic.ifPresent(t -> topicCache.put(qualifiedTopicName, cachedTopic(t)));
-  }
-
   @Override
   public Optional<CachedTopic> getTopic(String qualifiedTopicName) {
     return Optional.ofNullable(topicCache.get(qualifiedTopicName));
@@ -127,9 +93,4 @@ private CachedTopic cachedTopic(Topic topic) {
     return new CachedTopic(
         topic, metricsFacade, throughputRegistry, kafkaNamesMapper.toKafkaTopics(topic));
   }
-
-  private CachedTopic bannedTopic(Topic topic) {
-    return new CachedTopic(
-        topic, metricsFacade, throughputRegistry, kafkaNamesMapper.toKafkaTopics(topic), true);
-  }
 }

hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/config/FrontendConfiguration.java

@@ -3,7 +3,6 @@
 import jakarta.inject.Named;
 import java.time.Clock;
 import java.util.List;
-import org.apache.curator.framework.CuratorFramework;
 import org.springframework.boot.context.properties.EnableConfigurationProperties;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
@@ -12,7 +11,6 @@
 import pl.allegro.tech.hermes.domain.group.GroupRepository;
 import pl.allegro.tech.hermes.domain.notifications.InternalNotificationsBus;
 import pl.allegro.tech.hermes.domain.topic.TopicRepository;
-import pl.allegro.tech.hermes.frontend.blacklist.BlacklistZookeeperNotifyingCache;
 import pl.allegro.tech.hermes.frontend.buffer.BackupMessagesLoader;
 import pl.allegro.tech.hermes.frontend.buffer.PersistentBufferExtension;
 import pl.allegro.tech.hermes.frontend.cache.topic.NotificationBasedTopicsCache;
@@ -22,7 +20,6 @@
 import pl.allegro.tech.hermes.frontend.producer.BrokerMessageProducer;
 import pl.allegro.tech.hermes.frontend.validator.MessageValidators;
 import pl.allegro.tech.hermes.frontend.validator.TopicMessageValidator;
-import pl.allegro.tech.hermes.infrastructure.zookeeper.ZookeeperPaths;
 import pl.allegro.tech.hermes.schema.SchemaExistenceEnsurer;
 import pl.allegro.tech.hermes.schema.SchemaRepository;
 import pl.allegro.tech.hermes.tracker.frontend.Trackers;
@@ -43,12 +40,10 @@ public TopicsCache notificationBasedTopicsCache(
       TopicRepository topicRepository,
       MetricsFacade metricsFacade,
       ThroughputRegistry throughputRegistry,
-      KafkaNamesMapper kafkaNamesMapper,
-      BlacklistZookeeperNotifyingCache blacklistZookeeperNotifyingCache) {
+      KafkaNamesMapper kafkaNamesMapper) {
 
     return new NotificationBasedTopicsCache(
         internalNotificationsBus,
-        blacklistZookeeperNotifyingCache,
         groupRepository,
         topicRepository,
         metricsFacade,
@@ -86,12 +81,6 @@ public PersistentBufferExtension persistentBufferExtension(
         localMessageStorageProperties, clock, listeners, backupMessagesLoader, metricsFacade);
   }
 
-  @Bean(initMethod = "startup")
-  public BlacklistZookeeperNotifyingCache blacklistZookeeperNotifyingCache(
-      CuratorFramework curator, ZookeeperPaths zookeeperPaths) {
-    return new BlacklistZookeeperNotifyingCache(curator, zookeeperPaths);
-  }
-
   @Bean
   public BrokerListeners defaultBrokerListeners() {
     return new BrokerListeners();

hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/metric/CachedTopic.java

@@ -18,7 +18,6 @@ public class CachedTopic {
   private final Topic topic;
   private final KafkaTopics kafkaTopics;
   private final MetricsFacade metricsFacade;
-  private final boolean blacklisted;
 
   private final HermesTimer topicProducerLatencyTimer;
   private final HermesTimer globalProducerLatencyTimer;
@@ -47,19 +46,9 @@ public CachedTopic(
       MetricsFacade metricsFacade,
       ThroughputRegistry throughputRegistry,
       KafkaTopics kafkaTopics) {
-    this(topic, metricsFacade, throughputRegistry, kafkaTopics, false);
-  }
-
-  public CachedTopic(
-      Topic topic,
-      MetricsFacade metricsFacade,
-      ThroughputRegistry throughputRegistry,
-      KafkaTopics kafkaTopics,
-      boolean blacklisted) {
     this.topic = topic;
     this.kafkaTopics = kafkaTopics;
     this.metricsFacade = metricsFacade;
-    this.blacklisted = blacklisted;
 
     globalRequestMeter = metricsFacade.topics().topicGlobalRequestCounter();
     topicRequestMeter = metricsFacade.topics().topicRequestCounter(topic.getName());
@@ -104,10 +93,6 @@ public KafkaTopics getKafkaTopics() {
     return kafkaTopics;
   }
 
-  public boolean isBlacklisted() {
-    return blacklisted;
-  }
-
   public StartedTimersPair startProducerLatencyTimers() {
     return new StartedTimersPair(
         topicProducerLatencyTimer.time(), globalProducerLatencyTimer.time());

hermes-frontend/src/main/java/pl/allegro/tech/hermes/frontend/publishing/handlers/TopicHandler.java

@@ -2,7 +2,6 @@
 
 import static io.undertow.util.StatusCodes.INTERNAL_SERVER_ERROR;
 import static pl.allegro.tech.hermes.api.ErrorCode.AUTH_ERROR;
-import static pl.allegro.tech.hermes.api.ErrorCode.TOPIC_BLACKLISTED;
 import static pl.allegro.tech.hermes.api.ErrorCode.TOPIC_NOT_EXISTS;
 import static pl.allegro.tech.hermes.api.ErrorDescription.error;
 
@@ -67,16 +66,12 @@ private void onRequestValid(
     String topicName = exchange.getQueryParameters().get("qualifiedTopicName").getFirst();
     Optional<CachedTopic> maybeTopic = topicsCache.getTopic(topicName);
 
-    if (!maybeTopic.isPresent()) {
+    if (maybeTopic.isEmpty()) {
       unknownTopic(exchange, topicName, messageId);
       return;
     }
 
     CachedTopic cachedTopic = maybeTopic.get();
-    if (cachedTopic.isBlacklisted()) {
-      blacklistedTopic(exchange, topicName, messageId);
-      return;
-    }
 
     Topic topic = cachedTopic.getTopic();
     if (topic.isAuthEnabled() && !hasPermission(exchange, topic)) {
@@ -119,15 +114,6 @@ private void requestForbidden(
         exchange, error("Permission denied.", AUTH_ERROR), messageId, qualifiedTopicName);
   }
 
-  private void blacklistedTopic(
-      HttpServerExchange exchange, String qualifiedTopicName, String messageId) {
-    messageErrorProcessor.sendQuietly(
-        exchange,
-        error("Topic blacklisted: " + qualifiedTopicName, TOPIC_BLACKLISTED),
-        messageId,
-        qualifiedTopicName);
-  }
-
   // Default Undertow's response code (200) was changed in order to avoid situations in which
   // something wrong happens and Hermes-Frontend
   // does not publish message but return code 200