adding exception

beekeeper-cleanup/src/main/java/com/expediagroup/beekeeper/cleanup/validation/IcebergValidator.java

@@ -6,7 +6,7 @@
 import org.slf4j.LoggerFactory;
 
 import com.expediagroup.beekeeper.cleanup.metadata.CleanerClient;
-import com.expediagroup.beekeeper.core.error.BeekeeperException;
+import com.expediagroup.beekeeper.core.error.BeekeeperIcebergException;
 
 public class IcebergValidator {
 
@@ -21,7 +21,7 @@ public IcebergValidator(CleanerClient client) {
   /**
    * Beekeeper does not support Iceberg format right now. Iceberg tables in Hive Metastore do not store partition information,
    * so Beekeeper tries to clean up the entire table because that information is missing. This method checks if
-   * the table is an Iceberg table and throw BeekeeperException to stop the process.
+   * the table is an Iceberg table and throw IcebergTableFoundException to stop the process.
    *
    * @param databaseName
    * @param tableName
@@ -33,7 +33,7 @@ public void throwExceptionIfIceberg(String databaseName, String tableName) {
     String outputFormat = client.getOutputFormat(databaseName, tableName).toLowerCase();
 
     if (tableType.contains("iceberg") || format.contains("iceberg") || outputFormat.contains("iceberg")) {
-      throw new BeekeeperException("Iceberg tables are not currently supported in Beekeeper");
+      throw new BeekeeperIcebergException("Iceberg tables are not currently supported in Beekeeper");
     }
   }
 }

beekeeper-core/src/main/java/com/expediagroup/beekeeper/core/error/BeekeeperIcebergException.java

@@ -0,0 +1,33 @@
+/**
+ * Copyright (C) 2019 Expedia, Inc.
+ *
+ * Licensed 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 com.expediagroup.beekeeper.core.error;
+
+public class BeekeeperIcebergException extends BeekeeperException {
+
+  private static final long serialVersionUID = 1L;
+
+  public BeekeeperIcebergException(String message, Exception e) {
+    super(message, e);
+  }
+
+  public BeekeeperIcebergException(String message, Throwable e) {
+    super(message, e);
+  }
+
+  public BeekeeperIcebergException(String message) {
+    super(message);
+  }
+}

beekeeper-scheduler-apiary/src/main/java/com/expediagroup/beekeeper/scheduler/apiary/service/SchedulerApiary.java

@@ -22,12 +22,15 @@
 import java.util.List;
 import java.util.Optional;
 
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.annotation.Transactional;
 
 import com.expediagroup.beekeeper.cleanup.validation.IcebergValidator;
 import com.expediagroup.beekeeper.core.error.BeekeeperException;
+import com.expediagroup.beekeeper.core.error.BeekeeperIcebergException;
 import com.expediagroup.beekeeper.core.model.HousekeepingEntity;
 import com.expediagroup.beekeeper.core.model.LifecycleEventType;
 import com.expediagroup.beekeeper.scheduler.apiary.messaging.BeekeeperEventReader;
@@ -37,6 +40,8 @@
 @Component
 public class SchedulerApiary {
 
+  private static final Logger log = LoggerFactory.getLogger(SchedulerApiary.class);
+
   private final BeekeeperEventReader beekeeperEventReader;
   private final EnumMap<LifecycleEventType, SchedulerService> schedulerServiceMap;
   private final IcebergValidator icebergValidator;
@@ -65,6 +70,9 @@ public void scheduleBeekeeperEvent() {
         LifecycleEventType eventType = LifecycleEventType.valueOf(entity.getLifecycleType());
         SchedulerService scheduler = schedulerServiceMap.get(eventType);
         scheduler.scheduleForHousekeeping(entity);
+      } catch (BeekeeperIcebergException e) {
+        log.warn("Iceberg table are not supported in Beekeeper. Deleting message from queue", e);
+        beekeeperEventReader.delete(beekeeperEvent);
       } catch (Exception e) {
         throw new BeekeeperException(format(
             "Unable to schedule %s deletion for entity, this message will go back on the queue",