diff --git a/build.gradle b/build.gradle
index 42d9a496d..06e074d98 100644
--- a/build.gradle
+++ b/build.gradle
@@ -38,15 +38,18 @@ allprojects {
 
       // AWS Services - SDK v1 (for EC2, AutoScaling, STS)
       implementation 'com.amazonaws:aws-java-sdk-core:latest.release'
-      implementation 'com.amazonaws:aws-java-sdk-sns:latest.release'
       implementation 'com.amazonaws:aws-java-sdk-ec2:latest.release'
       implementation 'com.amazonaws:aws-java-sdk-autoscaling:latest.release'
       implementation 'com.amazonaws:aws-java-sdk-sts:latest.release'
 
-      // AWS Services - SDK v2 (for S3, SNS)
-      implementation platform('software.amazon.awssdk:bom:2.20.0')
+      // AWS Services - SDK v2 (for S3)
+      implementation platform('software.amazon.awssdk:bom:2.27.21')
       implementation 'software.amazon.awssdk:s3'
-      implementation 'software.amazon.awssdk:sts' // Needed for S3 role assumption
+        implementation 'software.amazon.awssdk:sts' // Needed for S3 role assumption
+        implementation("software.amazon.awssdk:imds")
+        implementation("software.amazon.awssdk:ec2")
+        implementation("software.amazon.awssdk:autoscaling")
+        implementation("software.amazon.awssdk:url-connection-client")
 
       implementation 'com.google.inject:guice:4.2.2'
       implementation 'com.google.inject.extensions:guice-servlet:4.2.2'
diff --git a/priam/src/main/java/com/netflix/priam/aws/AWSMembership.java b/priam/src/main/java/com/netflix/priam/aws/AWSMembership.java
index d2dd296bd..40d48c0c6 100644
--- a/priam/src/main/java/com/netflix/priam/aws/AWSMembership.java
+++ b/priam/src/main/java/com/netflix/priam/aws/AWSMembership.java
@@ -16,25 +16,26 @@
  */
 package com.netflix.priam.aws;
 
-import com.amazonaws.services.autoscaling.AmazonAutoScaling;
-import com.amazonaws.services.autoscaling.AmazonAutoScalingClientBuilder;
-import com.amazonaws.services.autoscaling.model.*;
-import com.amazonaws.services.autoscaling.model.Instance;
-import com.amazonaws.services.ec2.AmazonEC2;
-import com.amazonaws.services.ec2.AmazonEC2ClientBuilder;
-import com.amazonaws.services.ec2.model.*;
-import com.amazonaws.services.ec2.model.Filter;
 import com.google.common.collect.ImmutableSet;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.cred.ICredential;
 import com.netflix.priam.identity.IMembership;
 import com.netflix.priam.identity.config.InstanceInfo;
-import java.util.*;
-import javax.inject.Inject;
-import javax.inject.Named;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.autoscaling.AutoScalingClient;
+import software.amazon.awssdk.services.autoscaling.model.AutoScalingGroup;
+import software.amazon.awssdk.services.autoscaling.model.DescribeAutoScalingGroupsRequest;
+import software.amazon.awssdk.services.autoscaling.model.DescribeAutoScalingGroupsResponse;
+import software.amazon.awssdk.services.autoscaling.model.Instance;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+import java.util.ArrayList;
+import java.util.Arrays;
+import java.util.List;
 
 /**
  * Class to query amazon ASG for its members to provide - Number of valid nodes in the ASG - Number
@@ -45,41 +46,36 @@ public class AWSMembership implements IMembership {
     private final IConfiguration config;
     private final ICredential provider;
     private final InstanceInfo instanceInfo;
-    private final ICredential crossAccountProvider;
 
     @Inject
     public AWSMembership(
             IConfiguration config,
             ICredential provider,
-            @Named("awsec2roleassumption") ICredential crossAccountProvider,
             InstanceInfo instanceInfo) {
         this.config = config;
         this.provider = provider;
         this.instanceInfo = instanceInfo;
-        this.crossAccountProvider = crossAccountProvider;
     }
 
     @Override
     public ImmutableSet<String> getRacMembership() {
-        AmazonAutoScaling client = null;
-        try {
+        try (AutoScalingClient client = getAutoScalingClient()) {
             List<String> asgNames = new ArrayList<>();
             asgNames.add(instanceInfo.getAutoScalingGroup());
             asgNames.addAll(Arrays.asList(config.getSiblingASGNames().split("\\s*,\\s*")));
-            client = getAutoScalingClient();
             DescribeAutoScalingGroupsRequest asgReq =
-                    new DescribeAutoScalingGroupsRequest()
-                            .withAutoScalingGroupNames(
-                                    asgNames.toArray(new String[asgNames.size()]));
-            DescribeAutoScalingGroupsResult res = client.describeAutoScalingGroups(asgReq);
+                    DescribeAutoScalingGroupsRequest.builder()
+                            .autoScalingGroupNames(asgNames)
+                            .build();
+            DescribeAutoScalingGroupsResponse res = client.describeAutoScalingGroups(asgReq);
 
             ImmutableSet.Builder<String> instanceIds = ImmutableSet.builder();
-            for (AutoScalingGroup asg : res.getAutoScalingGroups()) {
-                for (Instance ins : asg.getInstances())
-                    if (!(ins.getLifecycleState().equalsIgnoreCase("Terminating")
-                            || ins.getLifecycleState().equalsIgnoreCase("shutting-down")
-                            || ins.getLifecycleState().equalsIgnoreCase("Terminated")))
-                        instanceIds.add(ins.getInstanceId());
+            for (AutoScalingGroup asg : res.autoScalingGroups()) {
+                for (Instance ins : asg.instances())
+                    if (!(ins.lifecycleStateAsString().equalsIgnoreCase("Terminating")
+                            || ins.lifecycleStateAsString().equalsIgnoreCase("shutting-down")
+                            || ins.lifecycleStateAsString().equalsIgnoreCase("Terminated")))
+                        instanceIds.add(ins.instanceId());
             }
             if (logger.isInfoEnabled()) {
                 logger.info(
@@ -90,63 +86,24 @@ public ImmutableSet<String> getRacMembership() {
                                 StringUtils.join(instanceIds, ",")));
             }
             return instanceIds.build();
-        } finally {
-            if (client != null) client.shutdown();
         }
     }
 
     /** Actual membership AWS source of truth... */
     @Override
     public int getRacMembershipSize() {
-        AmazonAutoScaling client = null;
-        try {
-            client = getAutoScalingClient();
+        try (AutoScalingClient client = getAutoScalingClient()) {
             DescribeAutoScalingGroupsRequest asgReq =
-                    new DescribeAutoScalingGroupsRequest()
-                            .withAutoScalingGroupNames(instanceInfo.getAutoScalingGroup());
-            DescribeAutoScalingGroupsResult res = client.describeAutoScalingGroups(asgReq);
+                    DescribeAutoScalingGroupsRequest.builder()
+                            .autoScalingGroupNames(instanceInfo.getAutoScalingGroup())
+                            .build();
+            DescribeAutoScalingGroupsResponse res = client.describeAutoScalingGroups(asgReq);
             int size = 0;
-            for (AutoScalingGroup asg : res.getAutoScalingGroups()) {
-                size += asg.getMaxSize();
+            for (AutoScalingGroup asg : res.autoScalingGroups()) {
+                size += asg.maxSize();
             }
             logger.info("Query on ASG returning {} instances", size);
             return size;
-        } finally {
-            if (client != null) client.shutdown();
-        }
-    }
-
-    @Override
-    public ImmutableSet<String> getCrossAccountRacMembership() {
-        AmazonAutoScaling client = null;
-        try {
-            List<String> asgNames = new ArrayList<>();
-            asgNames.add(instanceInfo.getAutoScalingGroup());
-            asgNames.addAll(Arrays.asList(config.getSiblingASGNames().split("\\s*,\\s*")));
-            client = getCrossAccountAutoScalingClient();
-            DescribeAutoScalingGroupsRequest asgReq =
-                    new DescribeAutoScalingGroupsRequest()
-                            .withAutoScalingGroupNames(
-                                    asgNames.toArray(new String[asgNames.size()]));
-            DescribeAutoScalingGroupsResult res = client.describeAutoScalingGroups(asgReq);
-
-            ImmutableSet.Builder<String> instanceIds = ImmutableSet.builder();
-            for (AutoScalingGroup asg : res.getAutoScalingGroups()) {
-                for (Instance ins : asg.getInstances())
-                    if (!(ins.getLifecycleState().equalsIgnoreCase("Terminating")
-                            || ins.getLifecycleState().equalsIgnoreCase("shutting-down")
-                            || ins.getLifecycleState().equalsIgnoreCase("Terminated")))
-                        instanceIds.add(ins.getInstanceId());
-            }
-            if (logger.isInfoEnabled()) {
-                logger.info(
-                        String.format(
-                                "Querying Amazon returned following instance in the cross-account ASG: %s --> %s",
-                                instanceInfo.getRac(), StringUtils.join(instanceIds, ",")));
-            }
-            return instanceIds.build();
-        } finally {
-            if (client != null) client.shutdown();
         }
     }
 
@@ -155,219 +112,10 @@ public int getRacCount() {
         return config.getRacs().size();
     }
 
-    private boolean isClassic() {
-        return instanceInfo.getInstanceEnvironment() == InstanceInfo.InstanceEnvironment.CLASSIC;
-    }
-
-    /**
-     * Adding peers' IPs as ingress to the running instance SG. The running instance could be in
-     * "classic" or "vpc"
-     */
-    public void addACL(Collection<String> listIPs, int from, int to) {
-        AmazonEC2 client = null;
-        try {
-            client = getEc2Client();
-            List<IpPermission> ipPermissions = new ArrayList<>();
-            ipPermissions.add(
-                    new IpPermission()
-                            .withFromPort(from)
-                            .withIpProtocol("tcp")
-                            .withIpRanges(listIPs)
-                            .withToPort(to));
-
-            if (isClassic()) {
-                client.authorizeSecurityGroupIngress(
-                        new AuthorizeSecurityGroupIngressRequest(
-                                config.getACLGroupName(), ipPermissions));
-                if (logger.isInfoEnabled()) {
-                    logger.info("Done adding ACL to classic: " + StringUtils.join(listIPs, ","));
-                }
-            } else {
-                // Adding peers' IPs as ingress to the running instance SG
-                AuthorizeSecurityGroupIngressRequest sgIngressRequest =
-                        new AuthorizeSecurityGroupIngressRequest()
-                                .withGroupId(getVpcGoupId())
-                                .withIpPermissions(ipPermissions);
-                // fetch SG group id for vpc account of the running instance.
-                int status =
-                        client.authorizeSecurityGroupIngress(sgIngressRequest)
-                                .getSdkHttpMetadata()
-                                .getHttpStatusCode();
-                if (status != 200) {
-                    logger.warn(
-                            "We might have too many ingress rules saw http status {} when updating",
-                            status);
-                }
-                if (logger.isInfoEnabled()) {
-                    logger.info("Done adding ACL to vpc: " + StringUtils.join(listIPs, ","));
-                }
-            }
-
-        } finally {
-            if (client != null) client.shutdown();
-        }
-    }
-
-    /*
-     * @return SG group id for a group name, vpc account of the running instance.
-     */
-    protected String getVpcGoupId() {
-        AmazonEC2 client = null;
-        try {
-            client = getEc2Client();
-            Filter nameFilter =
-                    new Filter().withName("group-name").withValues(config.getACLGroupName()); // SG
-            Filter vpcFilter = new Filter().withName("vpc-id").withValues(instanceInfo.getVpcId());
-
-            DescribeSecurityGroupsRequest req =
-                    new DescribeSecurityGroupsRequest().withFilters(nameFilter, vpcFilter);
-            DescribeSecurityGroupsResult result = client.describeSecurityGroups(req);
-            for (SecurityGroup group : result.getSecurityGroups()) {
-                logger.debug(
-                        "got group-id:{} for group-name:{},vpc-id:{}",
-                        group.getGroupId(),
-                        config.getACLGroupName(),
-                        instanceInfo.getVpcId());
-                return group.getGroupId();
-            }
-            logger.error(
-                    "unable to get group-id for group-name={} vpc-id={}",
-                    config.getACLGroupName(),
-                    instanceInfo.getVpcId());
-            return "";
-        } finally {
-            if (client != null) client.shutdown();
-        }
-    }
-
-    /** removes a iplist from the SG */
-    public void removeACL(Collection<String> listIPs, int from, int to) {
-        AmazonEC2 client = null;
-        try {
-            client = getEc2Client();
-            List<IpPermission> ipPermissions = new ArrayList<>();
-            ipPermissions.add(
-                    new IpPermission()
-                            .withFromPort(from)
-                            .withIpProtocol("tcp")
-                            .withIpRanges(listIPs)
-                            .withToPort(to));
-
-            if (isClassic()) {
-                client.revokeSecurityGroupIngress(
-                        new RevokeSecurityGroupIngressRequest(
-                                config.getACLGroupName(), ipPermissions));
-                if (logger.isInfoEnabled()) {
-                    logger.info(
-                            "Done removing from ACL within classic env for running instance: "
-                                    + StringUtils.join(listIPs, ","));
-                }
-            } else {
-                RevokeSecurityGroupIngressRequest req = new RevokeSecurityGroupIngressRequest();
-                // fetch SG group id for vpc account of the running instance.
-                req.withGroupId(getVpcGoupId());
-                // Adding peers' IPs as ingress to the running instance SG
-                client.revokeSecurityGroupIngress(req.withIpPermissions(ipPermissions));
-                if (logger.isInfoEnabled()) {
-                    logger.info(
-                            "Done removing from ACL within vpc env for running instance: "
-                                    + StringUtils.join(listIPs, ","));
-                }
-            }
-
-        } finally {
-            if (client != null) client.shutdown();
-        }
-    }
-
-    /** List SG ACL's */
-    public ImmutableSet<String> listACL(int from, int to) {
-        AmazonEC2 client = null;
-        try {
-            client = getEc2Client();
-            ImmutableSet.Builder<String> ipPermissions = ImmutableSet.builder();
-
-            if (isClassic()) {
-
-                DescribeSecurityGroupsRequest req =
-                        new DescribeSecurityGroupsRequest()
-                                .withGroupNames(
-                                        Collections.singletonList(config.getACLGroupName()));
-                DescribeSecurityGroupsResult result = client.describeSecurityGroups(req);
-                for (SecurityGroup group : result.getSecurityGroups())
-                    for (IpPermission perm : group.getIpPermissions())
-                        if (perm.getFromPort() == from && perm.getToPort() == to)
-                            ipPermissions.addAll(perm.getIpRanges());
-
-                logger.debug("Fetch current permissions for classic env of running instance");
-            } else {
-
-                Filter nameFilter =
-                        new Filter().withName("group-name").withValues(config.getACLGroupName());
-                String vpcid = instanceInfo.getVpcId();
-                if (vpcid == null || vpcid.isEmpty()) {
-                    throw new IllegalStateException(
-                            "vpcid is null even though instance is running in vpc.");
-                }
-
-                // only fetch SG for the vpc id of the running instance
-                Filter vpcFilter = new Filter().withName("vpc-id").withValues(vpcid);
-                DescribeSecurityGroupsRequest req =
-                        new DescribeSecurityGroupsRequest().withFilters(nameFilter, vpcFilter);
-                DescribeSecurityGroupsResult result = client.describeSecurityGroups(req);
-                for (SecurityGroup group : result.getSecurityGroups())
-                    for (IpPermission perm : group.getIpPermissions())
-                        if (perm.getFromPort() == from && perm.getToPort() == to)
-                            ipPermissions.addAll(perm.getIpRanges());
-
-                logger.debug("Fetch current permissions for vpc env of running instance");
-            }
-
-            return ipPermissions.build();
-        } finally {
-            if (client != null) client.shutdown();
-        }
-    }
-
-    @Override
-    public void expandRacMembership(int count) {
-        AmazonAutoScaling client = null;
-        try {
-            client = getAutoScalingClient();
-            DescribeAutoScalingGroupsRequest asgReq =
-                    new DescribeAutoScalingGroupsRequest()
-                            .withAutoScalingGroupNames(instanceInfo.getAutoScalingGroup());
-            DescribeAutoScalingGroupsResult res = client.describeAutoScalingGroups(asgReq);
-            AutoScalingGroup asg = res.getAutoScalingGroups().get(0);
-            UpdateAutoScalingGroupRequest ureq = new UpdateAutoScalingGroupRequest();
-            ureq.setAutoScalingGroupName(asg.getAutoScalingGroupName());
-            ureq.setMinSize(asg.getMinSize() + 1);
-            ureq.setMaxSize(asg.getMinSize() + 1);
-            ureq.setDesiredCapacity(asg.getMinSize() + 1);
-            client.updateAutoScalingGroup(ureq);
-        } finally {
-            if (client != null) client.shutdown();
-        }
-    }
-
-    protected AmazonAutoScaling getAutoScalingClient() {
-        return AmazonAutoScalingClientBuilder.standard()
-                .withCredentials(provider.getAwsCredentialProvider())
-                .withRegion(instanceInfo.getRegion())
-                .build();
-    }
-
-    protected AmazonAutoScaling getCrossAccountAutoScalingClient() {
-        return AmazonAutoScalingClientBuilder.standard()
-                .withCredentials(crossAccountProvider.getAwsCredentialProvider())
-                .withRegion(instanceInfo.getRegion())
-                .build();
-    }
-
-    protected AmazonEC2 getEc2Client() {
-        return AmazonEC2ClientBuilder.standard()
-                .withCredentials(provider.getAwsCredentialProvider())
-                .withRegion(instanceInfo.getRegion())
+    protected AutoScalingClient getAutoScalingClient() {
+        return AutoScalingClient.builder()
+                .region(Region.of(instanceInfo.getRegion()))
+                .credentialsProvider(provider.getAwsCredentialProvider())
                 .build();
     }
 }
diff --git a/priam/src/main/java/com/netflix/priam/aws/IAMCredential.java b/priam/src/main/java/com/netflix/priam/aws/IAMCredential.java
deleted file mode 100644
index d35674415..000000000
--- a/priam/src/main/java/com/netflix/priam/aws/IAMCredential.java
+++ /dev/null
@@ -1,46 +0,0 @@
-/*
- * Copyright 2013 Netflix, 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.netflix.priam.aws;
-
-import com.amazonaws.auth.AWSCredentials;
-import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.auth.InstanceProfileCredentialsProvider;
-import com.netflix.priam.cred.ICredential;
-
-public class IAMCredential implements ICredential {
-    private final InstanceProfileCredentialsProvider iamCredProvider;
-
-    public IAMCredential() {
-        this.iamCredProvider = InstanceProfileCredentialsProvider.getInstance();
-    }
-
-    public String getAccessKeyId() {
-        return iamCredProvider.getCredentials().getAWSAccessKeyId();
-    }
-
-    public String getSecretAccessKey() {
-        return iamCredProvider.getCredentials().getAWSSecretKey();
-    }
-
-    public AWSCredentials getCredentials() {
-        return iamCredProvider.getCredentials();
-    }
-
-    public AWSCredentialsProvider getAwsCredentialProvider() {
-        return iamCredProvider;
-    }
-}
diff --git a/priam/src/main/java/com/netflix/priam/aws/S3FileSystem.java b/priam/src/main/java/com/netflix/priam/aws/S3FileSystem.java
index bf52d2da1..b3269a138 100644
--- a/priam/src/main/java/com/netflix/priam/aws/S3FileSystem.java
+++ b/priam/src/main/java/com/netflix/priam/aws/S3FileSystem.java
@@ -18,7 +18,6 @@
 
 import com.google.common.annotations.VisibleForTesting;
 import com.google.common.base.Preconditions;
-import com.netflix.priam.aws.auth.IS3Credential;
 import com.netflix.priam.backup.AbstractBackupPath;
 import com.netflix.priam.backup.BackupRestoreException;
 import com.netflix.priam.backup.DynamicRateLimiter;
@@ -27,9 +26,9 @@
 import com.netflix.priam.compress.CompressionType;
 import com.netflix.priam.compress.ICompression;
 import com.netflix.priam.config.IConfiguration;
+import com.netflix.priam.cred.ICredential;
 import com.netflix.priam.identity.config.InstanceInfo;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.utils.BoundedExponentialRetryCallable;
 import com.netflix.priam.utils.ByteBufferInputStream;
 import com.netflix.priam.utils.SystemUtils;
@@ -70,15 +69,14 @@ public class S3FileSystem extends S3FileSystemBase {
 
     @Inject
     public S3FileSystem(
-            @Named("awss3roleassumption") IS3Credential cred,
+            @Named("awss3roleassumption") ICredential cred,
             Provider<AbstractBackupPath> pathProvider,
             ICompression compress,
             final IConfiguration config,
             BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr,
             InstanceInfo instanceInfo,
             DynamicRateLimiter dynamicRateLimiter) {
-        super(pathProvider, compress, config, backupMetrics, backupNotificationMgr);
+        super(pathProvider, compress, config, backupMetrics);
         s3Client =
                 S3Client.builder()
                         .credentialsProvider(cred.getAwsCredentialProvider())
diff --git a/priam/src/main/java/com/netflix/priam/aws/S3FileSystemBase.java b/priam/src/main/java/com/netflix/priam/aws/S3FileSystemBase.java
index f76e7f254..d046e4cce 100755
--- a/priam/src/main/java/com/netflix/priam/aws/S3FileSystemBase.java
+++ b/priam/src/main/java/com/netflix/priam/aws/S3FileSystemBase.java
@@ -13,7 +13,6 @@
  */
 package com.netflix.priam.aws;
 
-import com.google.common.collect.Lists;
 import com.google.common.util.concurrent.RateLimiter;
 import com.netflix.priam.backup.AbstractBackupPath;
 import com.netflix.priam.backup.AbstractFileSystem;
@@ -21,21 +20,20 @@
 import com.netflix.priam.compress.ICompression;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.scheduler.BlockingSubmitThreadPoolExecutor;
-import java.nio.file.Path;
-import java.util.Iterator;
-import java.util.List;
-import java.util.Optional;
-import java.util.concurrent.LinkedBlockingQueue;
-import java.util.stream.Collectors;
-import javax.inject.Provider;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import software.amazon.awssdk.core.exception.SdkException;
 import software.amazon.awssdk.services.s3.S3Client;
 import software.amazon.awssdk.services.s3.model.*;
 
+import javax.inject.Provider;
+import java.nio.file.Path;
+import java.util.Iterator;
+import java.util.List;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.stream.Collectors;
+
 public abstract class S3FileSystemBase extends AbstractFileSystem {
     private static final int MAX_CHUNKS = 9995; // 10K is AWS limit, minus a small buffer
     private static final Logger logger = LoggerFactory.getLogger(S3FileSystemBase.class);
@@ -50,9 +48,8 @@ public abstract class S3FileSystemBase extends AbstractFileSystem {
             Provider<AbstractBackupPath> pathProvider,
             ICompression compress,
             final IConfiguration config,
-            BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr) {
-        super(config, backupMetrics, backupNotificationMgr, pathProvider);
+            BackupMetrics backupMetrics) {
+        super(config, backupMetrics, pathProvider);
         this.compress = compress;
         this.config = config;
 
diff --git a/priam/src/main/java/com/netflix/priam/aws/auth/EC2RoleAssumptionCredential.java b/priam/src/main/java/com/netflix/priam/aws/auth/EC2RoleAssumptionCredential.java
index ef418dc7b..b1ed5fba1 100644
--- a/priam/src/main/java/com/netflix/priam/aws/auth/EC2RoleAssumptionCredential.java
+++ b/priam/src/main/java/com/netflix/priam/aws/auth/EC2RoleAssumptionCredential.java
@@ -13,19 +13,23 @@
  */
 package com.netflix.priam.aws.auth;
 
-import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.auth.STSAssumeRoleSessionCredentialsProvider;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.cred.ICredential;
 import com.netflix.priam.identity.config.InstanceInfo;
+import org.apache.commons.lang3.Validate;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+import software.amazon.awssdk.services.sts.StsClient;
+import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
+import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
+
 import javax.inject.Inject;
 
 public class EC2RoleAssumptionCredential implements ICredential {
-    private static final String AWS_ROLE_ASSUMPTION_SESSION_NAME = "AwsRoleAssumptionSession";
+    private static final String SESSION_NAME = "AwsRoleAssumptionSession";
     private final ICredential cred;
     private final IConfiguration config;
     private final InstanceInfo instanceInfo;
-    private AWSCredentialsProvider stsSessionCredentialsProvider;
+    private AwsCredentialsProvider stsSessionCredentialsProvider;
 
     @Inject
     public EC2RoleAssumptionCredential(
@@ -36,43 +40,25 @@ public EC2RoleAssumptionCredential(
     }
 
     @Override
-    public AWSCredentialsProvider getAwsCredentialProvider() {
-        if (this.config.isDualAccount() || this.stsSessionCredentialsProvider == null) {
+    public AwsCredentialsProvider getAwsCredentialProvider() {
+        if (this.stsSessionCredentialsProvider == null) {
             synchronized (this) {
                 if (this.stsSessionCredentialsProvider == null) {
-
-                    String roleArn;
-                    /**
-                     * Create the assumed IAM role based on the environment. For example, if the
-                     * current environment is VPC, then the assumed role is for EC2 classic, and
-                     * vice versa.
-                     */
-                    if (instanceInfo.getInstanceEnvironment()
-                            == InstanceInfo.InstanceEnvironment.CLASSIC) {
-                        roleArn = this.config.getClassicEC2RoleAssumptionArn();
-                        // Env is EC2 classic --> IAM assumed role for VPC created
-                    } else {
-                        roleArn = this.config.getVpcEC2RoleAssumptionArn();
-                        // Env is VPC --> IAM assumed role for EC2 classic created.
-                    }
-
-                    //
-                    if (roleArn == null || roleArn.isEmpty())
-                        throw new NullPointerException(
-                                "Role ARN is null or empty probably due to missing config entry");
-
-                    /**
-                     * Get handle to an implementation that uses AWS Security Token Service (STS) to
-                     * create temporary, short-lived session with explicit refresh for session/token
-                     * expiration.
-                     */
+                    String roleArn =
+                            instanceInfo.getInstanceEnvironment() == InstanceInfo.InstanceEnvironment.CLASSIC
+                                    ? this.config.getClassicEC2RoleAssumptionArn()
+                                    : this.config.getVpcEC2RoleAssumptionArn();
+                    Validate.notEmpty(roleArn, "roleArn is empty");
                     try {
+                        StsClient stsClient =
+                                StsClient.builder().credentialsProvider(cred.getAwsCredentialProvider()).build();
+                        AssumeRoleRequest assumeRoleRequest =
+                                AssumeRoleRequest.builder().roleArn(roleArn).roleSessionName(SESSION_NAME).build();
                         this.stsSessionCredentialsProvider =
-                                new STSAssumeRoleSessionCredentialsProvider(
-                                        this.cred.getAwsCredentialProvider(),
-                                        roleArn,
-                                        AWS_ROLE_ASSUMPTION_SESSION_NAME);
-
+                                StsAssumeRoleCredentialsProvider.builder()
+                                        .stsClient(stsClient)
+                                        .refreshRequest(assumeRoleRequest)
+                                        .build();
                     } catch (Exception ex) {
                         throw new IllegalStateException(
                                 "Exception in getting handle to AWS Security Token Service (STS).  Msg: "
@@ -82,7 +68,6 @@ public AWSCredentialsProvider getAwsCredentialProvider() {
                 }
             }
         }
-
         return this.stsSessionCredentialsProvider;
     }
 }
diff --git a/priam/src/main/java/com/netflix/priam/aws/auth/IS3Credential.java b/priam/src/main/java/com/netflix/priam/aws/auth/IS3Credential.java
deleted file mode 100755
index 27314c2df..000000000
--- a/priam/src/main/java/com/netflix/priam/aws/auth/IS3Credential.java
+++ /dev/null
@@ -1,24 +0,0 @@
-/**
- * Copyright 2017 Netflix, Inc.
- *
- * <p>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
- *
- * <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 com.netflix.priam.aws.auth;
-
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
-import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
-
-/*
- * Credentials specific to Amazon S3
- */
-public interface IS3Credential {
-    AwsCredentialsProvider getAwsCredentialProvider();
-}
diff --git a/priam/src/main/java/com/netflix/priam/aws/auth/S3InstanceCredential.java b/priam/src/main/java/com/netflix/priam/aws/auth/InstanceCredential.java
similarity index 81%
rename from priam/src/main/java/com/netflix/priam/aws/auth/S3InstanceCredential.java
rename to priam/src/main/java/com/netflix/priam/aws/auth/InstanceCredential.java
index bb7d6652f..0f92dff8a 100755
--- a/priam/src/main/java/com/netflix/priam/aws/auth/S3InstanceCredential.java
+++ b/priam/src/main/java/com/netflix/priam/aws/auth/InstanceCredential.java
@@ -13,18 +13,18 @@
  */
 package com.netflix.priam.aws.auth;
 
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
+import com.netflix.priam.cred.ICredential;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.InstanceProfileCredentialsProvider;
 
 /*
- * Provides credentials from the S3 instance.
+ * Provides credentials that work with SDK V2
  */
-public class S3InstanceCredential implements IS3Credential {
+public class InstanceCredential implements ICredential {
 
     private final InstanceProfileCredentialsProvider credentialsProvider;
 
-    public S3InstanceCredential() {
+    public InstanceCredential() {
         this.credentialsProvider = InstanceProfileCredentialsProvider.create();
     }
 
diff --git a/priam/src/main/java/com/netflix/priam/aws/auth/S3RoleAssumptionCredential.java b/priam/src/main/java/com/netflix/priam/aws/auth/S3RoleAssumptionCredential.java
index 4a40332a4..1232ee4b7 100755
--- a/priam/src/main/java/com/netflix/priam/aws/auth/S3RoleAssumptionCredential.java
+++ b/priam/src/main/java/com/netflix/priam/aws/auth/S3RoleAssumptionCredential.java
@@ -14,67 +14,52 @@
 package com.netflix.priam.aws.auth;
 
 import com.netflix.priam.config.IConfiguration;
-import javax.inject.Inject;
-import javax.inject.Named;
-import javax.inject.Singleton;
+import com.netflix.priam.cred.ICredential;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
-import software.amazon.awssdk.auth.credentials.AwsCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.services.sts.StsClient;
 import software.amazon.awssdk.services.sts.auth.StsAssumeRoleCredentialsProvider;
 import software.amazon.awssdk.services.sts.model.AssumeRoleRequest;
 
+import javax.inject.Inject;
+import javax.inject.Named;
+import javax.inject.Singleton;
+
 @Singleton
-public class S3RoleAssumptionCredential implements IS3Credential {
-    private static final String AWS_ROLE_ASSUMPTION_SESSION_NAME = "S3RoleAssumptionSession";
+public class S3RoleAssumptionCredential implements ICredential {
+    private static final String SESSION_NAME = "S3RoleAssumptionSession";
     private static final Logger logger = LoggerFactory.getLogger(S3RoleAssumptionCredential.class);
 
-    private final IS3Credential cred;
+    private final ICredential cred;
     private final IConfiguration config;
-    private AwsCredentialsProvider stsSessionCredentialsProvider;
+    private AwsCredentialsProvider credentialsProvider;
 
     @Inject
-    public S3RoleAssumptionCredential(
-            @Named("s3") IS3Credential cred, IConfiguration config) {
+    public S3RoleAssumptionCredential(@Named("s3") ICredential cred, IConfiguration config) {
         this.cred = cred;
         this.config = config;
     }
 
     public AwsCredentialsProvider getAwsCredentialProvider() {
-        if (this.stsSessionCredentialsProvider == null) {
+        if (this.credentialsProvider == null) {
             synchronized (this) {
-                if (this.stsSessionCredentialsProvider == null) {
-
-                    final String roleArn = this.config.getAWSRoleAssumptionArn();
-                    // IAM role created for bucket own by account "awsprodbackup"
+                if (this.credentialsProvider == null) {
+                    final String roleArn = config.getAWSRoleAssumptionArn();
                     if (roleArn == null || roleArn.isEmpty()) {
-                        logger.warn(
-                                "Role ARN is null or empty probably due to missing config entry. Falling back to instance level credentials");
-                        this.stsSessionCredentialsProvider = this.cred.getAwsCredentialProvider();
-                        // throw new NullPointerException("Role ARN is null or empty probably due to
-                        // missing config entry");
+                        logger.warn("Role ARN is empty due to missing config. Using instance level credentials");
+                        credentialsProvider = cred.getAwsCredentialProvider();
                     } else {
-                        // Get handle to an implementation that uses AWS Security Token Service
-                        // (STS) to create temporary, short-lived session with explicit refresh for
-                        // session/token expiration.
                         try {
-
-                            StsClient stsClient = StsClient.builder()
-                                    .credentialsProvider(cred.getAwsCredentialProvider())
-                                    .build();
-
-                            AssumeRoleRequest assumeRoleRequest = AssumeRoleRequest.builder()
-                                    .roleArn(roleArn)
-                                    .roleSessionName(AWS_ROLE_ASSUMPTION_SESSION_NAME)
-                                    .build();
-
-                            this.stsSessionCredentialsProvider =
+                            StsClient stsClient =
+                                    StsClient.builder().credentialsProvider(cred.getAwsCredentialProvider()).build();
+                            AssumeRoleRequest assumeRoleRequest =
+                                    AssumeRoleRequest.builder().roleArn(roleArn).roleSessionName(SESSION_NAME).build();
+                            credentialsProvider =
                                     StsAssumeRoleCredentialsProvider.builder()
                                             .stsClient(stsClient)
                                             .refreshRequest(assumeRoleRequest)
                                             .build();
-
                         } catch (Exception ex) {
                             throw new IllegalStateException(
                                     "Exception in getting handle to AWS Security Token Service (STS).  Msg: "
@@ -85,7 +70,6 @@ public AwsCredentialsProvider getAwsCredentialProvider() {
                 }
             }
         }
-
-        return this.stsSessionCredentialsProvider;
+        return credentialsProvider;
     }
 }
diff --git a/priam/src/main/java/com/netflix/priam/backup/AbstractFileSystem.java b/priam/src/main/java/com/netflix/priam/backup/AbstractFileSystem.java
index 2d38023b7..82abab697 100644
--- a/priam/src/main/java/com/netflix/priam/backup/AbstractFileSystem.java
+++ b/priam/src/main/java/com/netflix/priam/backup/AbstractFileSystem.java
@@ -26,28 +26,29 @@
 import com.google.common.util.concurrent.ListeningExecutorService;
 import com.google.common.util.concurrent.MoreExecutors;
 import com.netflix.priam.backup.AbstractBackupPath.BackupFileType;
-import com.netflix.priam.backupv2.IMetaProxy;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
-import com.netflix.priam.notification.UploadStatus;
 import com.netflix.priam.scheduler.BlockingSubmitThreadPoolExecutor;
 import com.netflix.priam.utils.BoundedExponentialRetryCallable;
 import com.netflix.spectator.api.patterns.PolledMeter;
-import java.io.File;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.time.Instant;
-import java.util.*;
-import java.util.concurrent.*;
-import javax.inject.Inject;
-import javax.inject.Provider;
 import org.apache.commons.collections4.iterators.TransformIterator;
 import org.apache.commons.io.FileUtils;
 import org.apache.commons.lang3.StringUtils;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
+import javax.inject.Inject;
+import javax.inject.Provider;
+import java.io.File;
+import java.nio.file.Path;
+import java.nio.file.Paths;
+import java.time.Instant;
+import java.util.Date;
+import java.util.Iterator;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.*;
+
 /**
  * This class is responsible for managing parallelism and orchestrating the upload and download, but
  * the subclasses actually implement the details of uploading a file.
@@ -62,7 +63,6 @@ public abstract class AbstractFileSystem implements IBackupFileSystem {
     private final Set<Path> tasksQueued;
     private final ListeningExecutorService fileUploadExecutor;
     private final ThreadPoolExecutor fileDownloadExecutor;
-    private final BackupNotificationMgr backupNotificationMgr;
 
     // This is going to be a write-thru cache containing the most frequently used items from remote
     // file system. This is to ensure that we don't make too many API calls to remote file system.
@@ -72,12 +72,10 @@ public abstract class AbstractFileSystem implements IBackupFileSystem {
     public AbstractFileSystem(
             IConfiguration configuration,
             BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr,
             Provider<AbstractBackupPath> pathProvider) {
         this.configuration = configuration;
         this.backupMetrics = backupMetrics;
         this.pathProvider = pathProvider;
-        this.backupNotificationMgr = backupNotificationMgr;
         this.objectCache =
                 CacheBuilder.newBuilder().maximumSize(configuration.getBackupQueueSize()).build();
         tasksQueued = new ConcurrentHashMap<>().newKeySet();
@@ -183,7 +181,6 @@ public AbstractBackupPath uploadAndDeleteInternal(
 
                 // Upload file if it not present at remote location.
                 if (path.getType() != BackupFileType.SST_V2 || path.isIncremental() || !checkObjectExists(remotePath)) {
-                    backupNotificationMgr.notify(path, UploadStatus.STARTED);
                     uploadedFileSize =
                             new BoundedExponentialRetryCallable<Long>(
                                     500 /* minSleep */, 10000 /* maxSleep */, retry) {
@@ -201,7 +198,6 @@ public Long retriableCall() throws Exception {
                     backupMetrics.recordUploadRate(uploadedFileSize);
                     backupMetrics.incrementValidUploads();
                     path.setCompressedFileSize(uploadedFileSize);
-                    backupNotificationMgr.notify(path, UploadStatus.SUCCESS);
                 } else {
                     // file is already uploaded to remote file system.
                     logger.info("File: {} already present on remoteFileSystem.", remotePath);
@@ -224,7 +220,6 @@ public Long retriableCall() throws Exception {
                         remotePath,
                         e.getMessage(),
                         e);
-                backupNotificationMgr.notify(path, UploadStatus.FAILED);
                 throw new BackupRestoreException(e.getMessage());
             } finally {
                 // Remove the task from the list so if we try to upload file ever again, we can.
diff --git a/priam/src/main/java/com/netflix/priam/backupv2/BackupVerificationTask.java b/priam/src/main/java/com/netflix/priam/backupv2/BackupVerificationTask.java
index e8fce7200..8879a5d5b 100644
--- a/priam/src/main/java/com/netflix/priam/backupv2/BackupVerificationTask.java
+++ b/priam/src/main/java/com/netflix/priam/backupv2/BackupVerificationTask.java
@@ -22,7 +22,6 @@
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.health.InstanceState;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.scheduler.CronTimer;
 import com.netflix.priam.scheduler.Task;
 import com.netflix.priam.scheduler.TaskTimer;
@@ -48,7 +47,6 @@ public class BackupVerificationTask extends Task {
     private final BackupVerification backupVerification;
     private final BackupMetrics backupMetrics;
     private final InstanceState instanceState;
-    private final BackupNotificationMgr backupNotificationMgr;
     private final SnapshotVerificationMarkerWriter snapshotVerificationMarkerWriter;
 
     @Inject
@@ -58,14 +56,12 @@ public BackupVerificationTask(
             BackupVerification backupVerification,
             BackupMetrics backupMetrics,
             InstanceState instanceState,
-            BackupNotificationMgr backupNotificationMgr,
             SnapshotVerificationMarkerWriter snapshotVerificationMarkerWriter) {
         super(configuration);
         this.backupRestoreConfig = backupRestoreConfig;
         this.backupVerification = backupVerification;
         this.backupMetrics = backupMetrics;
         this.instanceState = instanceState;
-        this.backupNotificationMgr = backupNotificationMgr;
         this.snapshotVerificationMarkerWriter = snapshotVerificationMarkerWriter;
     }
 
@@ -101,12 +97,6 @@ public void execute() throws Exception {
                                     snapshotLocation
                                             .subpath(1, snapshotLocation.getNameCount())
                                             .toString();
-                            logger.info(
-                                    "Sending {} message for backup: {}",
-                                    AbstractBackupPath.BackupFileType.SNAPSHOT_VERIFIED,
-                                    snapshotKey);
-                            backupNotificationMgr.notify(
-                                    snapshotKey, result.getStart().toInstant());
                             snapshotVerificationMarkerWriter.write(snapshotKey);
                         });
 
diff --git a/priam/src/main/java/com/netflix/priam/cli/StaticMembership.java b/priam/src/main/java/com/netflix/priam/cli/StaticMembership.java
index 2f1626737..eda0ff84e 100644
--- a/priam/src/main/java/com/netflix/priam/cli/StaticMembership.java
+++ b/priam/src/main/java/com/netflix/priam/cli/StaticMembership.java
@@ -20,7 +20,6 @@
 import com.netflix.priam.identity.IMembership;
 import java.io.FileInputStream;
 import java.io.IOException;
-import java.util.Collection;
 import java.util.Properties;
 import org.apache.cassandra.io.util.FileUtils;
 import org.slf4j.Logger;
@@ -66,11 +65,6 @@ public ImmutableSet<String> getRacMembership() {
         return racMembership;
     }
 
-    @Override
-    public ImmutableSet<String> getCrossAccountRacMembership() {
-        return null;
-    }
-
     @Override
     public int getRacMembershipSize() {
         if (racMembership == null) return 0;
@@ -81,18 +75,4 @@ public int getRacMembershipSize() {
     public int getRacCount() {
         return racCount;
     }
-
-    @Override
-    public void addACL(Collection<String> listIPs, int from, int to) {}
-
-    @Override
-    public void removeACL(Collection<String> listIPs, int from, int to) {}
-
-    @Override
-    public ImmutableSet<String> listACL(int from, int to) {
-        return null;
-    }
-
-    @Override
-    public void expandRacMembership(int count) {}
 }
diff --git a/priam/src/main/java/com/netflix/priam/config/PriamConfiguration.java b/priam/src/main/java/com/netflix/priam/config/PriamConfiguration.java
index 15fb15c69..bf172b686 100644
--- a/priam/src/main/java/com/netflix/priam/config/PriamConfiguration.java
+++ b/priam/src/main/java/com/netflix/priam/config/PriamConfiguration.java
@@ -548,11 +548,6 @@ public String getVpcEC2RoleAssumptionArn() {
         return config.get(PRIAM_PRE + ".vpc.roleassumption.arn");
     }
 
-    @Override
-    public boolean isDualAccount() {
-        return config.get(PRIAM_PRE + ".roleassumption.dualaccount", false);
-    }
-
     @Override
     public String getGcsServiceAccountId() {
         return config.get(PRIAM_PRE + ".gcs.service.acct.id");
diff --git a/priam/src/main/java/com/netflix/priam/cred/ClearCredential.java b/priam/src/main/java/com/netflix/priam/cred/ClearCredential.java
deleted file mode 100644
index 3d3a0ea6a..000000000
--- a/priam/src/main/java/com/netflix/priam/cred/ClearCredential.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/*
- * Copyright 2018 Netflix, 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.netflix.priam.cred;
-
-import com.amazonaws.auth.AWSCredentials;
-import com.amazonaws.auth.AWSCredentialsProvider;
-import com.amazonaws.auth.BasicAWSCredentials;
-import java.io.FileInputStream;
-import java.util.Properties;
-import org.apache.cassandra.io.util.FileUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This is a basic implementation of ICredentials. User should prefer to implement their own
- * versions for more secured access. This class requires clear AWS key and access.
- *
- * <p>Set the following properties in "conf/awscredntial.properties"
- */
-public class ClearCredential implements ICredential {
-    private static final Logger logger = LoggerFactory.getLogger(ClearCredential.class);
-    private static final String CRED_FILE = "/etc/awscredential.properties";
-    private final String AWS_ACCESS_ID;
-    private final String AWS_KEY;
-
-    public ClearCredential() {
-        FileInputStream fis = null;
-        try {
-            fis = new FileInputStream(CRED_FILE);
-            final Properties props = new Properties();
-            props.load(fis);
-            AWS_ACCESS_ID =
-                    props.getProperty("AWSACCESSID") != null
-                            ? props.getProperty("AWSACCESSID").trim()
-                            : "";
-            AWS_KEY = props.getProperty("AWSKEY") != null ? props.getProperty("AWSKEY").trim() : "";
-        } catch (Exception e) {
-            logger.error("Exception with credential file ", e);
-            throw new RuntimeException("Problem reading credential file. Cannot start.", e);
-        } finally {
-            FileUtils.closeQuietly(fis);
-        }
-    }
-
-    public AWSCredentialsProvider getAwsCredentialProvider() {
-        return new AWSCredentialsProvider() {
-            public AWSCredentials getCredentials() {
-                return new BasicAWSCredentials(AWS_ACCESS_ID, AWS_KEY);
-            }
-
-            public void refresh() {
-                // NOP
-            }
-        };
-    }
-}
diff --git a/priam/src/main/java/com/netflix/priam/cred/ICredential.java b/priam/src/main/java/com/netflix/priam/cred/ICredential.java
index 127ef1c48..cd5885649 100644
--- a/priam/src/main/java/com/netflix/priam/cred/ICredential.java
+++ b/priam/src/main/java/com/netflix/priam/cred/ICredential.java
@@ -16,12 +16,12 @@
  */
 package com.netflix.priam.cred;
 
-import com.amazonaws.auth.AWSCredentialsProvider;
-import com.google.inject.ImplementedBy;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 
 /** Credential file interface for services supporting Access ID and key authentication */
-@ImplementedBy(ClearCredential.class)
 public interface ICredential {
-    /** @return AWS Credential Provider object */
-    AWSCredentialsProvider getAwsCredentialProvider();
+    /**
+     * @return AWS Credential Provider object
+     */
+    AwsCredentialsProvider getAwsCredentialProvider();
 }
diff --git a/priam/src/main/java/com/netflix/priam/cryptography/pgp/PgpCredential.java b/priam/src/main/java/com/netflix/priam/cryptography/pgp/PgpCredential.java
index dcaa9af5a..b25f1dd1f 100755
--- a/priam/src/main/java/com/netflix/priam/cryptography/pgp/PgpCredential.java
+++ b/priam/src/main/java/com/netflix/priam/cryptography/pgp/PgpCredential.java
@@ -13,9 +13,10 @@
  */
 package com.netflix.priam.cryptography.pgp;
 
-import com.amazonaws.auth.AWSCredentialsProvider;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.cred.ICredentialGeneric;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
+
 import javax.inject.Inject;
 
 /*
@@ -32,7 +33,7 @@ public PgpCredential(IConfiguration config) {
     }
 
     @Override
-    public AWSCredentialsProvider getAwsCredentialProvider() {
+    public AwsCredentialsProvider getAwsCredentialProvider() {
         return null;
     }
 
diff --git a/priam/src/main/java/com/netflix/priam/defaultimpl/PriamGuiceModule.java b/priam/src/main/java/com/netflix/priam/defaultimpl/PriamGuiceModule.java
index 19ff006de..2919788c7 100644
--- a/priam/src/main/java/com/netflix/priam/defaultimpl/PriamGuiceModule.java
+++ b/priam/src/main/java/com/netflix/priam/defaultimpl/PriamGuiceModule.java
@@ -20,7 +20,6 @@
 import com.google.inject.name.Names;
 import com.netflix.priam.aws.S3FileSystem;
 import com.netflix.priam.aws.auth.EC2RoleAssumptionCredential;
-import com.netflix.priam.aws.auth.IS3Credential;
 import com.netflix.priam.aws.auth.S3RoleAssumptionCredential;
 import com.netflix.priam.backup.IBackupFileSystem;
 import com.netflix.priam.backupv2.IMetaProxy;
@@ -37,7 +36,7 @@ protected void configure() {
         bind(SchedulerFactory.class).to(StdSchedulerFactory.class).asEagerSingleton();
 
         bind(IBackupFileSystem.class).to(S3FileSystem.class);
-        bind(IS3Credential.class)
+        bind(ICredential.class)
                 .annotatedWith(Names.named("awss3roleassumption"))
                 .to(S3RoleAssumptionCredential.class);
         bind(ICredential.class)
diff --git a/priam/src/main/java/com/netflix/priam/identity/IMembership.java b/priam/src/main/java/com/netflix/priam/identity/IMembership.java
index ed6f72b18..59eafb326 100644
--- a/priam/src/main/java/com/netflix/priam/identity/IMembership.java
+++ b/priam/src/main/java/com/netflix/priam/identity/IMembership.java
@@ -37,49 +37,10 @@ public interface IMembership {
     /** @return Size of current RAC */
     int getRacMembershipSize();
 
-    /**
-     * Get a set of Instances in the cross-account but current RAC
-     *
-     * @return
-     */
-    ImmutableSet<String> getCrossAccountRacMembership();
-
     /**
      * Number of RACs
      *
      * @return
      */
     int getRacCount();
-
-    /**
-     * Add security group ACLs
-     *
-     * @param listIPs
-     * @param from
-     * @param to
-     */
-    void addACL(Collection<String> listIPs, int from, int to);
-
-    /**
-     * Remove security group ACLs
-     *
-     * @param listIPs
-     * @param from
-     * @param to
-     */
-    void removeACL(Collection<String> listIPs, int from, int to);
-
-    /**
-     * List all ACLs
-     *
-     * @return
-     */
-    ImmutableSet<String> listACL(int from, int to);
-
-    /**
-     * Expand the membership size by 1.
-     *
-     * @param count
-     */
-    void expandRacMembership(int count);
 }
diff --git a/priam/src/main/java/com/netflix/priam/identity/config/AWSInstanceInfo.java b/priam/src/main/java/com/netflix/priam/identity/config/AWSInstanceInfo.java
index 542496a56..db59f84cc 100644
--- a/priam/src/main/java/com/netflix/priam/identity/config/AWSInstanceInfo.java
+++ b/priam/src/main/java/com/netflix/priam/identity/config/AWSInstanceInfo.java
@@ -13,32 +13,39 @@
  */
 package com.netflix.priam.identity.config;
 
-import com.amazonaws.services.ec2.AmazonEC2;
-import com.amazonaws.services.ec2.AmazonEC2ClientBuilder;
-import com.amazonaws.services.ec2.model.*;
-import com.amazonaws.util.EC2MetadataUtils;
-import com.google.common.collect.ImmutableList;
-import com.google.common.collect.Lists;
 import com.netflix.priam.cred.ICredential;
 import com.netflix.priam.utils.RetryableCallable;
-import java.util.List;
-import javax.inject.Inject;
-import javax.inject.Singleton;
 import org.apache.commons.lang3.StringUtils;
-import org.codehaus.jettison.json.JSONObject;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
+import software.amazon.awssdk.imds.Ec2MetadataClient;
+import software.amazon.awssdk.regions.Region;
+import software.amazon.awssdk.services.ec2.Ec2Client;
+import software.amazon.awssdk.services.ec2.model.AvailabilityZone;
+import software.amazon.awssdk.services.ec2.model.DescribeInstancesRequest;
+import software.amazon.awssdk.services.ec2.model.Tag;
+
+import javax.inject.Inject;
+import javax.inject.Singleton;
+import java.util.List;
+import java.util.Optional;
+import java.util.stream.Collectors;
 
 @Singleton
 public class AWSInstanceInfo implements InstanceInfo {
     private static final Logger logger = LoggerFactory.getLogger(AWSInstanceInfo.class);
 
-    static final String PUBLIC_HOSTNAME_URL = "/latest/meta-data/public-hostname";
-    static final String LOCAL_HOSTNAME_URL = "/latest/meta-data/local-hostname";
-    static final String PUBLIC_HOSTIP_URL = "/latest/meta-data/public-ipv4";
-    static final String LOCAL_HOSTIP_URL = "/latest/meta-data/local-ipv4";
+    private static final String LATEST_METADATA = "/latest/meta-data/";
+    static final String PUBLIC_HOSTNAME_URL = LATEST_METADATA + "public-hostname";
+    static final String LOCAL_HOSTNAME_URL = LATEST_METADATA + "local-hostname";
+    static final String PUBLIC_HOSTIP_URL = LATEST_METADATA + "public-ipv4";
+    static final String LOCAL_HOSTIP_URL = LATEST_METADATA + "local-ipv4";
+    private static final String AVAILABILITY_ZONE = LATEST_METADATA + "placement/availability-zone";
+    private static final String INSTANCE_ID = LATEST_METADATA + "instance-id";
+    private static final String INSTANCE_TYPE = LATEST_METADATA + "instance-type";
+    private static final String REGION = LATEST_METADATA + "placement/region";
+    private static final String MAC = LATEST_METADATA + "mac";
 
-    private JSONObject identityDocument = null;
     private String privateIp;
     private String hostIP;
     private String rac;
@@ -47,7 +54,6 @@ public class AWSInstanceInfo implements InstanceInfo {
     private String instanceType;
     private String mac;
     private String region;
-    private String availabilityZone;
     private ICredential credential;
     private String vpcId;
     private InstanceEnvironment instanceEnvironment;
@@ -60,7 +66,7 @@ public AWSInstanceInfo(ICredential credential) {
     @Override
     public String getPrivateIP() {
         if (privateIp == null) {
-            privateIp = EC2MetadataUtils.getPrivateIpAddress();
+            privateIp = tryGetDataFromUrl(LOCAL_HOSTIP_URL);
         }
         return privateIp;
     }
@@ -68,32 +74,28 @@ public String getPrivateIP() {
     @Override
     public String getRac() {
         if (rac == null) {
-            rac = EC2MetadataUtils.getAvailabilityZone();
+            rac = tryGetDataFromUrl(AVAILABILITY_ZONE);
         }
         return rac;
     }
 
     @Override
     public List<String> getDefaultRacks() {
-        // Get the fist 3 available zones in the region
-        AmazonEC2 client =
-                AmazonEC2ClientBuilder.standard()
-                        .withCredentials(credential.getAwsCredentialProvider())
-                        .withRegion(getRegion())
-                        .build();
-        DescribeAvailabilityZonesResult res = client.describeAvailabilityZones();
-        List<String> zone = Lists.newArrayList();
-        for (AvailabilityZone reg : res.getAvailabilityZones()) {
-            if (reg.getState().equals("available")) zone.add(reg.getZoneName());
-            if (zone.size() == 3) break;
+        try (Ec2Client client = getEc2Client()) {
+            return client.describeAvailabilityZones()
+                    .availabilityZones()
+                    .stream()
+                    .filter(zone -> zone.stateAsString().equals("available"))
+                    .map(AvailabilityZone::zoneName)
+                    .limit(3)
+                    .collect(Collectors.toList());
         }
-        return ImmutableList.copyOf(zone);
     }
 
     @Override
     public String getInstanceId() {
         if (instanceId == null) {
-            instanceId = EC2MetadataUtils.getInstanceId();
+            instanceId = tryGetDataFromUrl(INSTANCE_ID);
         }
         return instanceId;
     }
@@ -101,14 +103,14 @@ public String getInstanceId() {
     @Override
     public String getInstanceType() {
         if (instanceType == null) {
-            instanceType = EC2MetadataUtils.getInstanceType();
+            instanceType = tryGetDataFromUrl(INSTANCE_TYPE);
         }
         return instanceType;
     }
 
     private String getMac() {
         if (mac == null) {
-            mac = EC2MetadataUtils.getNetworkInterfaces().get(0).getMacAddress();
+            mac = tryGetDataFromUrl(MAC);
         }
         return mac;
     }
@@ -116,51 +118,40 @@ private String getMac() {
     @Override
     public String getRegion() {
         if (region == null) {
-            region = EC2MetadataUtils.getEC2InstanceRegion();
+            region = tryGetDataFromUrl(REGION);
         }
         return region;
     }
 
     @Override
     public String getVpcId() {
-        String nacId = getMac();
-        if (StringUtils.isEmpty(nacId)) return null;
-
-        if (vpcId == null)
-            try {
-                vpcId = EC2MetadataUtils.getNetworkInterfaces().get(0).getVpcId();
-            } catch (Exception e) {
-                logger.info(
-                        "Vpc id does not exist for running instance, not fatal as running instance maybe not be in vpc.  Msg: {}",
-                        e.getLocalizedMessage());
+        if (vpcId == null) {
+            String mac = getMac();
+            if (!StringUtils.isEmpty(mac)) {
+                vpcId = tryGetDataFromUrl(LATEST_METADATA + "network/interfaces/macs/" + mac + "/vpc-id");
             }
-
+        }
         return vpcId;
     }
 
     @Override
     public String getAutoScalingGroup() {
-        final AmazonEC2 client =
-                AmazonEC2ClientBuilder.standard()
-                        .withCredentials(credential.getAwsCredentialProvider())
-                        .withRegion(getRegion())
-                        .build();
-        try {
+        try (Ec2Client client = getEc2Client()) {
             return new RetryableCallable<String>(15, 30000) {
                 public String retriableCall() throws IllegalStateException {
                     DescribeInstancesRequest desc =
-                            new DescribeInstancesRequest().withInstanceIds(getInstanceId());
-                    DescribeInstancesResult res = client.describeInstances(desc);
-
-                    for (Reservation resr : res.getReservations()) {
-                        for (Instance ins : resr.getInstances()) {
-                            for (com.amazonaws.services.ec2.model.Tag tag : ins.getTags()) {
-                                if (tag.getKey().equals("aws:autoscaling:groupName"))
-                                    return tag.getValue();
-                            }
-                        }
+                            DescribeInstancesRequest.builder().instanceIds(getInstanceId()).build();
+                    Optional<Tag> matchingTag =
+                            client.describeInstances(desc)
+                                    .reservations()
+                                    .stream()
+                                    .flatMap(res -> res.instances().stream())
+                                    .flatMap(instance -> instance.tags().stream())
+                                    .filter(tag -> tag.key().equals("aws:autoscaling:groupName"))
+                                    .findFirst();
+                    if (matchingTag.isPresent()) {
+                        return matchingTag.get().value();
                     }
-
                     throw new IllegalStateException("Couldn't determine ASG name");
                 }
             }.call();
@@ -173,8 +164,7 @@ public String retriableCall() throws IllegalStateException {
     @Override
     public InstanceEnvironment getInstanceEnvironment() {
         if (instanceEnvironment == null) {
-            instanceEnvironment =
-                    (getVpcId() == null) ? InstanceEnvironment.CLASSIC : InstanceEnvironment.VPC;
+            instanceEnvironment = (getVpcId() == null) ? InstanceEnvironment.CLASSIC : InstanceEnvironment.VPC;
         }
         return instanceEnvironment;
     }
@@ -183,8 +173,7 @@ public InstanceEnvironment getInstanceEnvironment() {
     public String getHostname() {
         if (hostName == null) {
             String publicHostName = tryGetDataFromUrl(PUBLIC_HOSTNAME_URL);
-            hostName =
-                    publicHostName == null ? tryGetDataFromUrl(LOCAL_HOSTNAME_URL) : publicHostName;
+            hostName = publicHostName == null ? tryGetDataFromUrl(LOCAL_HOSTNAME_URL) : publicHostName;
         }
         return hostName;
     }
@@ -198,9 +187,17 @@ public String getHostIP() {
         return hostIP;
     }
 
+    private Ec2Client getEc2Client() {
+        return Ec2Client.builder()
+                .region(Region.of(getRegion()))
+                .credentialsProvider(credential.getAwsCredentialProvider())
+                .build();
+    }
+
     String tryGetDataFromUrl(String url) {
-        try {
-            return EC2MetadataUtils.getData(url);
+        try (Ec2MetadataClient client = Ec2MetadataClient.create()) {
+            // trim()? -> V2 client sometimes appends newlines
+            return client.get(url).asString().trim();
         } catch (Exception e) {
             return null;
         }
diff --git a/priam/src/main/java/com/netflix/priam/identity/token/TokenRetriever.java b/priam/src/main/java/com/netflix/priam/identity/token/TokenRetriever.java
index e368a1a4b..7e1fa6aa1 100644
--- a/priam/src/main/java/com/netflix/priam/identity/token/TokenRetriever.java
+++ b/priam/src/main/java/com/netflix/priam/identity/token/TokenRetriever.java
@@ -128,7 +128,7 @@ public PriamInstance grabExistingToken() throws Exception {
             public PriamInstance retriableCall() throws Exception {
                 logger.info("Trying to grab an existing token");
                 sleeper.sleep(new Random().nextInt(5000) + 10000);
-                Set<String> racInstanceIds = getRacInstanceIds();
+                Set<String> racInstanceIds = membership.getRacMembership();
                 ImmutableSet<PriamInstance> allIds = factory.getAllIds(config.getAppName());
                 List<PriamInstance> instances =
                         allIds.stream()
@@ -350,13 +350,6 @@ private Optional<PriamInstance> findInstance(ImmutableSet<PriamInstance> instanc
                 .findFirst();
     }
 
-    private Set<String> getRacInstanceIds() { // TODO(CASS-1986)
-        ImmutableSet<String> racMembership = membership.getRacMembership();
-        return config.isDualAccount()
-                ? Sets.union(membership.getCrossAccountRacMembership(), racMembership)
-                : racMembership;
-    }
-
     private boolean isNew(PriamInstance instance) {
         return instance.getInstanceId().equals(NEW_SLOT);
     }
diff --git a/priam/src/main/java/com/netflix/priam/notification/AWSSnsNotificationService.java b/priam/src/main/java/com/netflix/priam/notification/AWSSnsNotificationService.java
deleted file mode 100644
index 564647ff5..000000000
--- a/priam/src/main/java/com/netflix/priam/notification/AWSSnsNotificationService.java
+++ /dev/null
@@ -1,107 +0,0 @@
-/*
- * Copyright 2017 Netflix, 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.netflix.priam.notification;
-
-import com.amazonaws.services.sns.AmazonSNS;
-import com.amazonaws.services.sns.AmazonSNSClient;
-import com.amazonaws.services.sns.model.MessageAttributeValue;
-import com.amazonaws.services.sns.model.PublishRequest;
-import com.amazonaws.services.sns.model.PublishResult;
-import com.netflix.priam.aws.IAMCredential;
-import com.netflix.priam.config.IConfiguration;
-import com.netflix.priam.identity.config.InstanceInfo;
-import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.utils.BoundedExponentialRetryCallable;
-import java.util.Map;
-import javax.inject.Inject;
-import javax.inject.Singleton;
-import org.apache.commons.lang3.StringUtils;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/*
- * A single, persisted, connection to Amazon SNS.
- */
-@Singleton
-public class AWSSnsNotificationService implements INotificationService {
-    private static final Logger logger = LoggerFactory.getLogger(AWSSnsNotificationService.class);
-
-    private final IConfiguration configuration;
-    private final AmazonSNS snsClient;
-    private final BackupMetrics backupMetrics;
-
-    @Inject
-    public AWSSnsNotificationService(
-            IConfiguration config,
-            IAMCredential iamCredential,
-            BackupMetrics backupMetrics,
-            InstanceInfo instanceInfo) {
-        this.configuration = config;
-        this.backupMetrics = backupMetrics;
-        String ec2_region = instanceInfo.getRegion();
-        snsClient =
-                AmazonSNSClient.builder()
-                        .withCredentials(iamCredential.getAwsCredentialProvider())
-                        .withRegion(ec2_region)
-                        .build();
-    }
-
-    @Override
-    public void notify(
-            final String msg, final Map<String, MessageAttributeValue> messageAttributes) {
-        // e.g. arn:aws:sns:eu-west-1:1234:eu-west-1-cass-sample-backup
-        final String topic_arn = this.configuration.getBackupNotificationTopicArn();
-        if (!configuration.enableBackupNotification() || StringUtils.isEmpty(topic_arn)) {
-            return;
-        }
-
-        PublishResult publishResult;
-        try {
-            publishResult =
-                    new BoundedExponentialRetryCallable<PublishResult>() {
-                        @Override
-                        public PublishResult retriableCall() throws Exception {
-                            PublishRequest publishRequest =
-                                    new PublishRequest(topic_arn, msg)
-                                            .withMessageAttributes(messageAttributes);
-                            return snsClient.publish(publishRequest);
-                        }
-                    }.call();
-
-        } catch (Exception e) {
-            logger.error(
-                    String.format(
-                            "Exhausted retries.  Publishing notification metric for failure and moving on.  Failed msg to publish: %s",
-                            msg),
-                    e);
-            backupMetrics.incrementSnsNotificationFailure();
-            return;
-        }
-
-        // If here, message was published.  As a extra validation, ensure we have a msg id
-        String publishedMsgId = publishResult.getMessageId();
-        if (publishedMsgId == null || publishedMsgId.isEmpty()) {
-            backupMetrics.incrementSnsNotificationFailure();
-            return;
-        }
-
-        backupMetrics.incrementSnsNotificationSuccess();
-        if (logger.isTraceEnabled()) {
-            logger.trace("Published msg:  {} aws sns messageId - {}", msg, publishedMsgId);
-        }
-    }
-}
diff --git a/priam/src/main/java/com/netflix/priam/notification/BackupNotificationMgr.java b/priam/src/main/java/com/netflix/priam/notification/BackupNotificationMgr.java
deleted file mode 100644
index 10f7373d2..000000000
--- a/priam/src/main/java/com/netflix/priam/notification/BackupNotificationMgr.java
+++ /dev/null
@@ -1,174 +0,0 @@
-/**
- * Copyright 2017 Netflix, Inc.
- *
- * <p>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
- *
- * <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 com.netflix.priam.notification;
-
-import com.amazonaws.services.sns.model.MessageAttributeValue;
-import com.netflix.priam.backup.AbstractBackupPath;
-import com.netflix.priam.backup.BackupRestoreException;
-import com.netflix.priam.config.IBackupRestoreConfig;
-import com.netflix.priam.config.IConfiguration;
-import com.netflix.priam.identity.InstanceIdentity;
-import com.netflix.priam.identity.config.InstanceInfo;
-import java.time.Instant;
-import java.util.*;
-import javax.inject.Inject;
-import org.apache.commons.lang3.StringUtils;
-import org.codehaus.jettison.json.JSONException;
-import org.codehaus.jettison.json.JSONObject;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * A means to nofity interested party(ies) of an uploaded file, success or failed.
- *
- * <p>Created by vinhn on 10/30/16.
- */
-public class BackupNotificationMgr {
-
-    private static final Logger logger = LoggerFactory.getLogger(BackupNotificationMgr.class);
-    private final IConfiguration config;
-    private final IBackupRestoreConfig backupRestoreConfig;
-    private final INotificationService notificationService;
-    private final InstanceInfo instanceInfo;
-    private final InstanceIdentity instanceIdentity;
-    private final Set<AbstractBackupPath.BackupFileType> notifiedBackupFileTypesSet;
-    private String notifiedBackupFileTypes;
-
-    @Inject
-    public BackupNotificationMgr(
-            IConfiguration config,
-            IBackupRestoreConfig backupRestoreConfig,
-            INotificationService notificationService,
-            InstanceInfo instanceInfo,
-            InstanceIdentity instanceIdentity) {
-        this.config = config;
-        this.backupRestoreConfig = backupRestoreConfig;
-        this.notificationService = notificationService;
-        this.instanceInfo = instanceInfo;
-        this.instanceIdentity = instanceIdentity;
-        this.notifiedBackupFileTypesSet = new HashSet<>();
-        this.notifiedBackupFileTypes = "";
-    }
-
-    public void notify(String remotePath, Instant snapshotInstant) {
-        JSONObject jsonObject = new JSONObject();
-        try {
-            jsonObject.put("s3bucketname", this.config.getBackupPrefix());
-            jsonObject.put("s3clustername", config.getAppName());
-            jsonObject.put("s3namespace", remotePath);
-            jsonObject.put("region", instanceInfo.getRegion());
-            jsonObject.put("rack", instanceInfo.getRac());
-            jsonObject.put("token", instanceIdentity.getInstance().getToken());
-            jsonObject.put(
-                    "backuptype", AbstractBackupPath.BackupFileType.SNAPSHOT_VERIFIED.name());
-            jsonObject.put("snapshotInstant", snapshotInstant);
-            // SNS Attributes for filtering messages. Cluster name and backup file type.
-            Map<String, MessageAttributeValue> messageAttributes = getMessageAttributes(jsonObject);
-
-            this.notificationService.notify(jsonObject.toString(), messageAttributes);
-        } catch (JSONException exception) {
-            logger.error(
-                    "JSON exception during generation of notification for snapshot verification: {}. path: {}, time: {}",
-                    remotePath,
-                    snapshotInstant,
-                    exception.getLocalizedMessage());
-        }
-    }
-
-    private Map<String, MessageAttributeValue> getMessageAttributes(JSONObject message)
-            throws JSONException {
-        Map<String, MessageAttributeValue> attributes = new HashMap<>();
-        attributes.put("s3clustername", toStringAttribute(message.getString("s3clustername")));
-        attributes.put("backuptype", toStringAttribute(message.getString("backuptype")));
-        for (String attr : backupRestoreConfig.getBackupNotificationAdditionalMessageAttrs()) {
-            if (message.has(attr)) {
-                attributes.put(attr, toStringAttribute(String.valueOf(message.get(attr))));
-            }
-        }
-        return attributes;
-    }
-
-    private MessageAttributeValue toStringAttribute(String value) {
-        return new MessageAttributeValue().withDataType("String").withStringValue(value);
-    }
-
-    public void notify(AbstractBackupPath abp, UploadStatus uploadStatus) {
-        JSONObject jsonObject = new JSONObject();
-        try {
-            Set<AbstractBackupPath.BackupFileType> updatedNotifiedBackupFileTypeSet =
-                    getUpdatedNotifiedBackupFileTypesSet(this.notifiedBackupFileTypes);
-            if (updatedNotifiedBackupFileTypeSet.isEmpty()
-                    || updatedNotifiedBackupFileTypeSet.contains(abp.getType())) {
-                jsonObject.put("s3bucketname", this.config.getBackupPrefix());
-                jsonObject.put("s3clustername", abp.getClusterName());
-                jsonObject.put("s3namespace", abp.getRemotePath());
-                jsonObject.put("keyspace", abp.getKeyspace());
-                jsonObject.put("cf", abp.getColumnFamily());
-                jsonObject.put("region", abp.getRegion());
-                jsonObject.put("rack", instanceInfo.getRac());
-                jsonObject.put("token", abp.getToken());
-                jsonObject.put("filename", abp.getFileName());
-                jsonObject.put("uncompressfilesize", abp.getSize());
-                jsonObject.put("compressfilesize", abp.getCompressedFileSize());
-                jsonObject.put("backuptype", abp.getType().name());
-                jsonObject.put("uploadstatus", uploadStatus.name().toLowerCase());
-                jsonObject.put("compression", abp.getCompression().name());
-                jsonObject.put("encryption", abp.getEncryption().name());
-                jsonObject.put("isincremental", abp.isIncremental());
-
-                // SNS Attributes for filtering messages. Cluster name and backup file type.
-                Map<String, MessageAttributeValue> messageAttributes =
-                        getMessageAttributes(jsonObject);
-
-                this.notificationService.notify(jsonObject.toString(), messageAttributes);
-            } else {
-                logger.debug(
-                        "BackupFileType {} is not in the list of notified component types {}",
-                        abp.getType().name(),
-                        StringUtils.join(notifiedBackupFileTypesSet, ", "));
-            }
-        } catch (JSONException exception) {
-            logger.error(
-                    "JSON exception during generation of notification for upload {}.  Local file {}. Ignoring to continue with rest of backup.  Msg: {}",
-                    uploadStatus,
-                    abp.getFileName(),
-                    exception.getLocalizedMessage());
-        }
-    }
-
-    private Set<AbstractBackupPath.BackupFileType> getUpdatedNotifiedBackupFileTypesSet(
-            String notifiedBackupFileTypes) {
-        String propertyValue = this.backupRestoreConfig.getBackupNotifyComponentIncludeList();
-        if (!notifiedBackupFileTypes.equals(propertyValue)) {
-            logger.info(
-                    String.format(
-                            "Notified BackupFileTypes changed from %s to %s",
-                            this.notifiedBackupFileTypes, propertyValue));
-            this.notifiedBackupFileTypesSet.clear();
-            this.notifiedBackupFileTypes =
-                    this.backupRestoreConfig.getBackupNotifyComponentIncludeList();
-            if (!StringUtils.isBlank(this.notifiedBackupFileTypes)) {
-                for (String s : this.notifiedBackupFileTypes.split(",")) {
-                    try {
-                        AbstractBackupPath.BackupFileType backupFileType =
-                                AbstractBackupPath.BackupFileType.valueOf(s.trim());
-                        notifiedBackupFileTypesSet.add(backupFileType);
-                    } catch (IllegalArgumentException ignored) {
-                    }
-                }
-            }
-        }
-        return Collections.unmodifiableSet(this.notifiedBackupFileTypesSet);
-    }
-}
diff --git a/priam/src/main/java/com/netflix/priam/notification/INotificationService.java b/priam/src/main/java/com/netflix/priam/notification/INotificationService.java
deleted file mode 100644
index befeb9a43..000000000
--- a/priam/src/main/java/com/netflix/priam/notification/INotificationService.java
+++ /dev/null
@@ -1,30 +0,0 @@
-/**
- * Copyright 2017 Netflix, Inc.
- *
- * <p>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
- *
- * <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 com.netflix.priam.notification;
-
-import com.amazonaws.services.sns.model.MessageAttributeValue;
-import com.google.inject.ImplementedBy;
-import java.util.Map;
-
-/** Service to notify of a message. Created by vinhn on 11/3/16. */
-@ImplementedBy(AWSSnsNotificationService.class)
-public interface INotificationService {
-    /**
-     * Notify the message.
-     *
-     * @param msg Message that needs to be notified
-     * @param messageAttributes Message attributes to be used while sending the message.
-     */
-    void notify(String msg, Map<String, MessageAttributeValue> messageAttributes);
-}
diff --git a/priam/src/main/java/com/netflix/priam/notification/UploadStatus.java b/priam/src/main/java/com/netflix/priam/notification/UploadStatus.java
deleted file mode 100644
index 158a42ce3..000000000
--- a/priam/src/main/java/com/netflix/priam/notification/UploadStatus.java
+++ /dev/null
@@ -1,7 +0,0 @@
-package com.netflix.priam.notification;
-
-public enum UploadStatus {
-    STARTED,
-    SUCCESS,
-    FAILED
-}
diff --git a/priam/src/main/java/com/netflix/priam/resources/BackupServletV2.java b/priam/src/main/java/com/netflix/priam/resources/BackupServletV2.java
index bc099a2a4..4fda0241f 100644
--- a/priam/src/main/java/com/netflix/priam/resources/BackupServletV2.java
+++ b/priam/src/main/java/com/netflix/priam/resources/BackupServletV2.java
@@ -19,16 +19,20 @@
 
 import com.fasterxml.jackson.databind.ObjectMapper;
 import com.netflix.priam.backup.*;
-import com.netflix.priam.backupv2.BackupTTLTask;
-import com.netflix.priam.backupv2.BackupV2Service;
-import com.netflix.priam.backupv2.IMetaProxy;
-import com.netflix.priam.backupv2.SnapshotMetaTask;
 import com.netflix.priam.backupv2.*;
 import com.netflix.priam.config.IConfiguration;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.utils.DateUtil;
 import com.netflix.priam.utils.DateUtil.DateRange;
 import com.netflix.priam.utils.GsonJsonSerializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import javax.inject.Inject;
+import javax.inject.Named;
+import javax.inject.Provider;
+import javax.ws.rs.*;
+import javax.ws.rs.core.MediaType;
+import javax.ws.rs.core.Response;
 import java.time.Instant;
 import java.time.temporal.ChronoUnit;
 import java.util.HashMap;
@@ -36,14 +40,6 @@
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
-import javax.inject.Inject;
-import javax.inject.Named;
-import javax.inject.Provider;
-import javax.ws.rs.*;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 
 /** Created by aagrawal on 1/16/19. */
 @Path("/v2/backup")
@@ -60,7 +56,6 @@ public class BackupServletV2 {
     private final IMetaProxy metaProxy;
     private final Provider<AbstractBackupPath> pathProvider;
     private final BackupV2Service backupService;
-    private final BackupNotificationMgr backupNotificationMgr;
     private final IConfiguration config;
     private final DirectorySize directorySize;
     private final SnapshotVerificationMarkerWriter snapshotVerificationMarkerWriter;
@@ -77,7 +72,6 @@ public BackupServletV2(
             @Named("v2") IMetaProxy metaV2Proxy,
             Provider<AbstractBackupPath> pathProvider,
             BackupV2Service backupService,
-            BackupNotificationMgr backupNotificationMgr,
             IConfiguration config,
             DirectorySize directorySize,
             SnapshotVerificationMarkerWriter snapshotVerificationMarkerWriter) {
@@ -89,7 +83,6 @@ public BackupServletV2(
         this.metaProxy = metaV2Proxy;
         this.pathProvider = pathProvider;
         this.backupService = backupService;
-        this.backupNotificationMgr = backupNotificationMgr;
         this.config = config;
         this.directorySize = directorySize;
         this.snapshotVerificationMarkerWriter = snapshotVerificationMarkerWriter;
@@ -156,19 +149,7 @@ public Response validateV2SnapshotByDate(
                     .entity("No valid meta found for provided time range")
                     .build();
         }
-
-        // Send notification for any verified backups. This is useful in one-off backup consumption
-        // by downward dependencies.
-        // Side-effect: It may send notification for already verified snapshot i.e. duplicate
-        // message may be sent.
-        logger.info(
-                "Sending {} message for backup: {}",
-                AbstractBackupPath.BackupFileType.SNAPSHOT_VERIFIED,
-                result.get().remotePath);
-
-        backupNotificationMgr.notify(result.get().remotePath, result.get().snapshotInstant);
         snapshotVerificationMarkerWriter.write(result.get().remotePath);
-
         return Response.ok(result.get().toString()).build();
     }
 
diff --git a/priam/src/main/java/com/netflix/priam/resources/SecurityGroupAdmin.java b/priam/src/main/java/com/netflix/priam/resources/SecurityGroupAdmin.java
deleted file mode 100644
index f1a54c198..000000000
--- a/priam/src/main/java/com/netflix/priam/resources/SecurityGroupAdmin.java
+++ /dev/null
@@ -1,70 +0,0 @@
-/**
- * Copyright 2017 Netflix, Inc.
- *
- * <p>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
- *
- * <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 com.netflix.priam.resources;
-
-import com.netflix.priam.identity.IMembership;
-import java.util.Collections;
-import javax.inject.Inject;
-import javax.ws.rs.*;
-import javax.ws.rs.core.MediaType;
-import javax.ws.rs.core.Response;
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
-
-/**
- * This http endpoint allows direct updates (adding/removing) (CIDR) IP addresses and port ranges to
- * the security group for this app.
- */
-@Path("/v1/secgroup")
-@Produces(MediaType.TEXT_PLAIN)
-public class SecurityGroupAdmin {
-    private static final Logger log = LoggerFactory.getLogger(SecurityGroupAdmin.class);
-    private static final String CIDR_TAG = "/32";
-    private final IMembership membership;
-
-    @Inject
-    public SecurityGroupAdmin(IMembership membership) {
-        this.membership = membership;
-    }
-
-    @POST
-    public Response addACL(
-            @QueryParam("ip") String ipAddr,
-            @QueryParam("fromPort") int fromPort,
-            @QueryParam("toPort") int toPort) {
-        if (!ipAddr.endsWith(CIDR_TAG)) ipAddr += CIDR_TAG;
-        try {
-            membership.addACL(Collections.singletonList(ipAddr), fromPort, toPort);
-        } catch (Exception e) {
-            log.error("Error while trying to add an ACL to a security group", e);
-            return Response.serverError().build();
-        }
-        return Response.ok().build();
-    }
-
-    @DELETE
-    public Response removeACL(
-            @QueryParam("ip") String ipAddr,
-            @QueryParam("fromPort") int fromPort,
-            @QueryParam("toPort") int toPort) {
-        if (!ipAddr.endsWith(CIDR_TAG)) ipAddr += CIDR_TAG;
-        try {
-            membership.removeACL(Collections.singletonList(ipAddr), fromPort, toPort);
-        } catch (Exception e) {
-            log.error("Error while trying to remove an ACL to a security group", e);
-            return Response.serverError().build();
-        }
-        return Response.ok().build();
-    }
-}
diff --git a/priam/src/test/java/com/netflix/priam/backup/BRTestModule.java b/priam/src/test/java/com/netflix/priam/backup/BRTestModule.java
index e7d488d50..03a1868ab 100644
--- a/priam/src/test/java/com/netflix/priam/backup/BRTestModule.java
+++ b/priam/src/test/java/com/netflix/priam/backup/BRTestModule.java
@@ -20,7 +20,6 @@
 import com.google.inject.AbstractModule;
 import com.google.inject.Scopes;
 import com.google.inject.name.Names;
-import com.netflix.priam.aws.auth.IS3Credential;
 import com.netflix.priam.aws.auth.S3RoleAssumptionCredential;
 import com.netflix.priam.backupv2.IMetaProxy;
 import com.netflix.priam.backupv2.MetaV2Proxy;
@@ -42,13 +41,14 @@
 import com.netflix.priam.utils.Sleeper;
 import com.netflix.spectator.api.DefaultRegistry;
 import com.netflix.spectator.api.Registry;
+import org.junit.Ignore;
+import org.quartz.SchedulerFactory;
+import org.quartz.impl.StdSchedulerFactory;
+
 import java.time.Clock;
 import java.time.Instant;
 import java.time.ZoneId;
 import java.util.Collections;
-import org.junit.Ignore;
-import org.quartz.SchedulerFactory;
-import org.quartz.impl.StdSchedulerFactory;
 
 @Ignore
 public class BRTestModule extends AbstractModule {
@@ -68,11 +68,11 @@ protected void configure() {
         bind(IBackupFileSystem.class).to(FakeBackupFileSystem.class).in(Scopes.SINGLETON);
         bind(Sleeper.class).to(FakeSleeper.class);
 
-        bind(IS3Credential.class)
+        bind(ICredential.class)
                 .annotatedWith(Names.named("s3"))
                 .to(FakeS3Credential.class)
                 .in(Scopes.SINGLETON);
-        bind(IS3Credential.class)
+        bind(ICredential.class)
                 .annotatedWith(Names.named("awss3roleassumption"))
                 .to(S3RoleAssumptionCredential.class);
 
diff --git a/priam/src/test/java/com/netflix/priam/backup/FakeBackupFileSystem.java b/priam/src/test/java/com/netflix/priam/backup/FakeBackupFileSystem.java
index 4472796f9..b095bbcc6 100644
--- a/priam/src/test/java/com/netflix/priam/backup/FakeBackupFileSystem.java
+++ b/priam/src/test/java/com/netflix/priam/backup/FakeBackupFileSystem.java
@@ -19,7 +19,6 @@
 
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import java.io.IOException;
 import java.nio.file.Files;
 import java.nio.file.Path;
@@ -43,9 +42,8 @@ public class FakeBackupFileSystem extends AbstractFileSystem {
     public FakeBackupFileSystem(
             IConfiguration configuration,
             BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr,
             Provider<AbstractBackupPath> pathProvider) {
-        super(configuration, backupMetrics, backupNotificationMgr, pathProvider);
+        super(configuration, backupMetrics, pathProvider);
     }
 
     public void setupTest(List<String> files) {
diff --git a/priam/src/test/java/com/netflix/priam/backup/FakeCredentials.java b/priam/src/test/java/com/netflix/priam/backup/FakeCredentials.java
index b73fe6e10..42cfe4b15 100644
--- a/priam/src/test/java/com/netflix/priam/backup/FakeCredentials.java
+++ b/priam/src/test/java/com/netflix/priam/backup/FakeCredentials.java
@@ -17,11 +17,11 @@
 
 package com.netflix.priam.backup;
 
-import com.amazonaws.auth.AWSCredentialsProvider;
 import com.netflix.priam.cred.ICredential;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 
 public class FakeCredentials implements ICredential {
-    public AWSCredentialsProvider getAwsCredentialProvider() {
+    public AwsCredentialsProvider getAwsCredentialProvider() {
         // TODO Auto-generated method stub
         return null;
     }
diff --git a/priam/src/test/java/com/netflix/priam/backup/FakeNullCredential.java b/priam/src/test/java/com/netflix/priam/backup/FakeNullCredential.java
index 5e99fbd46..88cbb1b04 100644
--- a/priam/src/test/java/com/netflix/priam/backup/FakeNullCredential.java
+++ b/priam/src/test/java/com/netflix/priam/backup/FakeNullCredential.java
@@ -17,11 +17,11 @@
 
 package com.netflix.priam.backup;
 
-import com.amazonaws.auth.AWSCredentialsProvider;
 import com.netflix.priam.cred.ICredential;
+import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 
 class FakeNullCredential implements ICredential {
-    public AWSCredentialsProvider getAwsCredentialProvider() {
+    public AwsCredentialsProvider getAwsCredentialProvider() {
         // TODO Auto-generated method stub
         return null;
     }
diff --git a/priam/src/test/java/com/netflix/priam/backup/FakeS3Credential.java b/priam/src/test/java/com/netflix/priam/backup/FakeS3Credential.java
index 5fbba49be..2e87ab635 100644
--- a/priam/src/test/java/com/netflix/priam/backup/FakeS3Credential.java
+++ b/priam/src/test/java/com/netflix/priam/backup/FakeS3Credential.java
@@ -17,13 +17,13 @@
 
 package com.netflix.priam.backup;
 
-import com.netflix.priam.aws.auth.IS3Credential;
+import com.netflix.priam.cred.ICredential;
 import software.amazon.awssdk.auth.credentials.AwsBasicCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentials;
 import software.amazon.awssdk.auth.credentials.AwsCredentialsProvider;
 import software.amazon.awssdk.auth.credentials.StaticCredentialsProvider;
 
-public class FakeS3Credential implements IS3Credential {
+public class FakeS3Credential implements ICredential {
     private final AwsCredentialsProvider credentialsProvider;
 
     public FakeS3Credential() {
diff --git a/priam/src/test/java/com/netflix/priam/backup/NullBackupFileSystem.java b/priam/src/test/java/com/netflix/priam/backup/NullBackupFileSystem.java
index 0eaa45724..63fe1589c 100644
--- a/priam/src/test/java/com/netflix/priam/backup/NullBackupFileSystem.java
+++ b/priam/src/test/java/com/netflix/priam/backup/NullBackupFileSystem.java
@@ -19,7 +19,6 @@
 
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import java.nio.file.Path;
 import java.time.Instant;
 import java.util.Collections;
@@ -34,9 +33,8 @@ public class NullBackupFileSystem extends AbstractFileSystem {
     public NullBackupFileSystem(
             IConfiguration configuration,
             BackupMetrics backupMetrics,
-            BackupNotificationMgr backupNotificationMgr,
             Provider<AbstractBackupPath> pathProvider) {
-        super(configuration, backupMetrics, backupNotificationMgr, pathProvider);
+        super(configuration, backupMetrics, pathProvider);
     }
 
     public void shutdown() {
diff --git a/priam/src/test/java/com/netflix/priam/backup/TestAbstractFileSystem.java b/priam/src/test/java/com/netflix/priam/backup/TestAbstractFileSystem.java
index bd666f1d2..89ed1cd6a 100644
--- a/priam/src/test/java/com/netflix/priam/backup/TestAbstractFileSystem.java
+++ b/priam/src/test/java/com/netflix/priam/backup/TestAbstractFileSystem.java
@@ -21,7 +21,6 @@
 import com.google.inject.Injector;
 import com.netflix.priam.config.IConfiguration;
 import com.netflix.priam.merics.BackupMetrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.utils.BackupFileUtils;
 import java.io.File;
 import java.nio.file.Path;
@@ -49,7 +48,6 @@ public class TestAbstractFileSystem {
     private Injector injector;
     private IConfiguration configuration;
     private BackupMetrics backupMetrics;
-    private BackupNotificationMgr backupNotificationMgr;
     private FailureFileSystem failureFileSystem;
     private MyFileSystem myFileSystem;
 
@@ -59,21 +57,18 @@ public void setBackupMetrics() {
 
         if (configuration == null) configuration = injector.getInstance(IConfiguration.class);
 
-        if (backupNotificationMgr == null)
-            backupNotificationMgr = injector.getInstance(BackupNotificationMgr.class);
-
         backupMetrics = injector.getInstance(BackupMetrics.class);
         Provider<AbstractBackupPath> pathProvider = injector.getProvider(AbstractBackupPath.class);
 
         if (failureFileSystem == null)
             failureFileSystem =
                     new FailureFileSystem(
-                            configuration, backupMetrics, backupNotificationMgr, pathProvider);
+                            configuration, backupMetrics, pathProvider);
 
         if (myFileSystem == null)
             myFileSystem =
                     new MyFileSystem(
-                            configuration, backupMetrics, backupNotificationMgr, pathProvider);
+                            configuration, backupMetrics, pathProvider);
 
         BackupFileUtils.cleanupDir(Paths.get(configuration.getDataFileLocation()));
     }
@@ -276,9 +271,8 @@ class FailureFileSystem extends NullBackupFileSystem {
         public FailureFileSystem(
                 IConfiguration configuration,
                 BackupMetrics backupMetrics,
-                BackupNotificationMgr backupNotificationMgr,
                 Provider<AbstractBackupPath> pathProvider) {
-            super(configuration, backupMetrics, backupNotificationMgr, pathProvider);
+            super(configuration, backupMetrics, pathProvider);
         }
 
         @Override
@@ -306,9 +300,8 @@ class MyFileSystem extends NullBackupFileSystem {
         public MyFileSystem(
                 IConfiguration configuration,
                 BackupMetrics backupMetrics,
-                BackupNotificationMgr backupNotificationMgr,
                 Provider<AbstractBackupPath> pathProvider) {
-            super(configuration, backupMetrics, backupNotificationMgr, pathProvider);
+            super(configuration, backupMetrics, pathProvider);
         }
 
         @Override
diff --git a/priam/src/test/java/com/netflix/priam/backupv2/TestBackupVerificationTask.java b/priam/src/test/java/com/netflix/priam/backupv2/TestBackupVerificationTask.java
index bdb091186..54a2bf1fc 100644
--- a/priam/src/test/java/com/netflix/priam/backupv2/TestBackupVerificationTask.java
+++ b/priam/src/test/java/com/netflix/priam/backupv2/TestBackupVerificationTask.java
@@ -24,7 +24,6 @@
 import com.netflix.priam.backup.*;
 import com.netflix.priam.health.InstanceState;
 import com.netflix.priam.merics.Metrics;
-import com.netflix.priam.notification.BackupNotificationMgr;
 import com.netflix.priam.scheduler.UnsupportedTypeException;
 import com.netflix.priam.utils.DateUtil.DateRange;
 import com.netflix.spectator.api.Counter;
@@ -45,12 +44,10 @@ public class TestBackupVerificationTask {
     @Inject private BackupVerificationTask backupVerificationService;
     private Counter badVerifications;
     @Mocked private BackupVerification backupVerification;
-    @Mocked private BackupNotificationMgr backupNotificationMgr;
 
     @Before
     public void setUp() {
         new MockBackupVerification();
-        new MockBackupNotificationMgr();
         Injector injector = Guice.createInjector(new BRTestModule());
         injector.injectMembers(this);
         badVerifications =
@@ -91,8 +88,6 @@ public Optional<BackupVerificationResult> verifyLatestBackup(
         }
     }
 
-    private static final class MockBackupNotificationMgr extends MockUp<BackupNotificationMgr> {}
-
     @Test
     public void throwError() {
         MockBackupVerification.shouldThrow(true);
@@ -106,12 +101,6 @@ public void validBackups() throws Exception {
         MockBackupVerification.setVerifiedBackups(getRecentlyValidatedMetadata());
         backupVerificationService.execute();
         Truth.assertThat(badVerifications.count()).isEqualTo(0);
-        new Verifications() {
-            {
-                backupNotificationMgr.notify(anyString, (Instant) any);
-                times = 1;
-            }
-        };
     }
 
     @Test
@@ -128,12 +117,6 @@ public void previouslyVerifiedBackups() throws Exception {
         MockBackupVerification.setVerifiedBackups(getPreviouslyValidatedMetadata());
         backupVerificationService.execute();
         Truth.assertThat(badVerifications.count()).isEqualTo(0);
-        new Verifications() {
-            {
-                backupNotificationMgr.notify(anyString, (Instant) any);
-                times = 0;
-            }
-        };
     }
 
     @Test
@@ -142,12 +125,6 @@ public void noBackups() throws Exception {
         MockBackupVerification.setVerifiedBackups();
         backupVerificationService.execute();
         Truth.assertThat(badVerifications.count()).isEqualTo(1);
-        new Verifications() {
-            {
-                backupNotificationMgr.notify(anyString, (Instant) any);
-                maxTimes = 0;
-            }
-        };
     }
 
     @Test
@@ -165,11 +142,6 @@ public void testRestoreMode(@Mocked InstanceState state) throws Exception {
                 backupVerification.verifyBackupsInRange((DateRange) any);
                 maxTimes = 0;
             }
-
-            {
-                backupNotificationMgr.notify(anyString, (Instant) any);
-                maxTimes = 0;
-            }
         };
     }
 
diff --git a/priam/src/test/java/com/netflix/priam/identity/FakeMembership.java b/priam/src/test/java/com/netflix/priam/identity/FakeMembership.java
index b320442d4..4cfb64c35 100644
--- a/priam/src/test/java/com/netflix/priam/identity/FakeMembership.java
+++ b/priam/src/test/java/com/netflix/priam/identity/FakeMembership.java
@@ -18,7 +18,7 @@
 package com.netflix.priam.identity;
 
 import com.google.common.collect.ImmutableSet;
-import java.util.Collection;
+
 import java.util.HashSet;
 import java.util.List;
 import java.util.Set;
@@ -38,11 +38,6 @@ public ImmutableSet<String> getRacMembership() {
         return instances;
     }
 
-    @Override
-    public ImmutableSet<String> getCrossAccountRacMembership() {
-        return null;
-    }
-
     @Override
     public int getRacMembershipSize() {
         return 3;
@@ -52,25 +47,4 @@ public int getRacMembershipSize() {
     public int getRacCount() {
         return 3;
     }
-
-    @Override
-    public void addACL(Collection<String> listIPs, int from, int to) {
-        acl.addAll(listIPs);
-    }
-
-    @Override
-    public void removeACL(Collection<String> listIPs, int from, int to) {
-        acl.removeAll(listIPs);
-    }
-
-    @Override
-    public ImmutableSet<String> listACL(int from, int to) {
-        return ImmutableSet.copyOf(acl);
-    }
-
-    @Override
-    public void expandRacMembership(int count) {
-        // TODO Auto-generated method stub
-
-    }
 }
diff --git a/priam/src/test/java/com/netflix/priam/notification/TestBackupNotificationMgr.java b/priam/src/test/java/com/netflix/priam/notification/TestBackupNotificationMgr.java
deleted file mode 100644
index cf4f05a75..000000000
--- a/priam/src/test/java/com/netflix/priam/notification/TestBackupNotificationMgr.java
+++ /dev/null
@@ -1,306 +0,0 @@
-package com.netflix.priam.notification;
-
-import com.amazonaws.services.sns.model.MessageAttributeValue;
-import com.google.inject.Guice;
-import com.google.inject.Injector;
-import com.netflix.priam.backup.AbstractBackupPath;
-import com.netflix.priam.backup.BRTestModule;
-import com.netflix.priam.config.IBackupRestoreConfig;
-import com.netflix.priam.config.IConfiguration;
-import java.nio.file.Path;
-import java.nio.file.Paths;
-import java.text.ParseException;
-import java.time.Instant;
-import java.util.Map;
-import javax.inject.Provider;
-import mockit.Capturing;
-import mockit.Expectations;
-import mockit.Mocked;
-import mockit.Verifications;
-import org.junit.Before;
-import org.junit.Test;
-
-public class TestBackupNotificationMgr {
-    private Injector injector;
-    private BackupNotificationMgr backupNotificationMgr;
-    private Provider<AbstractBackupPath> abstractBackupPathProvider;
-    private IConfiguration configuration;
-
-    @Before
-    public void setUp() {
-        if (injector == null) {
-            injector = Guice.createInjector(new BRTestModule());
-        }
-
-        if (backupNotificationMgr == null) {
-            backupNotificationMgr = injector.getInstance(BackupNotificationMgr.class);
-        }
-
-        if (abstractBackupPathProvider == null) {
-            abstractBackupPathProvider = injector.getProvider(AbstractBackupPath.class);
-        }
-    }
-
-    @Test
-    public void testNotificationNonEmptyFilter(
-            @Mocked IBackupRestoreConfig backupRestoreConfig,
-            @Capturing INotificationService notificationService)
-            throws ParseException {
-        new Expectations() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                result = "SNAPSHOT_VERIFIED, META_V2";
-                maxTimes = 2;
-            }
-        };
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-        Path path =
-                Paths.get(
-                        "fakeDataLocation",
-                        "fakeKeyspace",
-                        "fakeColumnFamily",
-                        "fakeBackup",
-                        "fakeData.db");
-        AbstractBackupPath abstractBackupPath = abstractBackupPathProvider.get();
-        abstractBackupPath.parseLocal(path.toFile(), AbstractBackupPath.BackupFileType.META_V2);
-        backupNotificationMgr.notify(abstractBackupPath, UploadStatus.STARTED);
-        new Verifications() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                maxTimes = 2;
-            }
-
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-    }
-
-    @Test
-    public void testNoNotificationsNonEmptyFilter(
-            @Mocked IBackupRestoreConfig backupRestoreConfig,
-            @Capturing INotificationService notificationService)
-            throws ParseException {
-        new Expectations() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                result = "META_V2";
-                maxTimes = 2;
-            }
-        };
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 0;
-            }
-        };
-        Path path =
-                Paths.get(
-                        "fakeDataLocation",
-                        "fakeKeyspace",
-                        "fakeColumnFamily",
-                        "fakeBackup",
-                        "fakeData.db");
-        AbstractBackupPath abstractBackupPath = abstractBackupPathProvider.get();
-        abstractBackupPath.parseLocal(path.toFile(), AbstractBackupPath.BackupFileType.SST_V2);
-        backupNotificationMgr.notify(abstractBackupPath, UploadStatus.STARTED);
-        new Verifications() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                maxTimes = 2;
-            }
-
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 0;
-            }
-        };
-    }
-
-    @Test
-    public void testNotificationsEmptyFilter(
-            @Mocked IBackupRestoreConfig backupRestoreConfig,
-            @Capturing INotificationService notificationService)
-            throws ParseException {
-        new Expectations() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                result = "";
-                maxTimes = 1;
-            }
-        };
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-        Path path =
-                Paths.get(
-                        "fakeDataLocation",
-                        "fakeKeyspace",
-                        "fakeColumnFamily",
-                        "fakeBackup",
-                        "fakeData.db");
-        AbstractBackupPath abstractBackupPath = abstractBackupPathProvider.get();
-        abstractBackupPath.parseLocal(path.toFile(), AbstractBackupPath.BackupFileType.SST_V2);
-        backupNotificationMgr.notify(abstractBackupPath, UploadStatus.STARTED);
-        new Verifications() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                maxTimes = 1;
-            }
-
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-    }
-
-    @Test
-    public void testNotificationsInvalidFilter(
-            @Mocked IBackupRestoreConfig backupRestoreConfig,
-            @Capturing INotificationService notificationService)
-            throws ParseException {
-        new Expectations() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                result = "SOME_FAKE_FILE_TYPE_1, SOME_FAKE_FILE_TYPE_2";
-                maxTimes = 2;
-            }
-        };
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-        Path path =
-                Paths.get(
-                        "fakeDataLocation",
-                        "fakeKeyspace",
-                        "fakeColumnFamily",
-                        "fakeBackup",
-                        "fakeData.db");
-        AbstractBackupPath abstractBackupPath = abstractBackupPathProvider.get();
-        abstractBackupPath.parseLocal(path.toFile(), AbstractBackupPath.BackupFileType.SST_V2);
-        backupNotificationMgr.notify(abstractBackupPath, UploadStatus.STARTED);
-        new Verifications() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                maxTimes = 2;
-            }
-
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-    }
-
-    @Test
-    public void testNotificationsPartiallyValidFilter(
-            @Mocked IBackupRestoreConfig backupRestoreConfig,
-            @Capturing INotificationService notificationService)
-            throws ParseException {
-        new Expectations() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                result = "SOME_FAKE_FILE_TYPE_1, SOME_FAKE_FILE_TYPE_2, META_V2";
-                maxTimes = 2;
-            }
-        };
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-        Path path =
-                Paths.get(
-                        "fakeDataLocation",
-                        "fakeKeyspace",
-                        "fakeColumnFamily",
-                        "fakeBackup",
-                        "fakeData.db");
-        AbstractBackupPath abstractBackupPath = abstractBackupPathProvider.get();
-        abstractBackupPath.parseLocal(path.toFile(), AbstractBackupPath.BackupFileType.META_V2);
-        backupNotificationMgr.notify(abstractBackupPath, UploadStatus.STARTED);
-        new Verifications() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                maxTimes = 2;
-            }
-
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-    }
-
-    @Test
-    public void testNoNotificationsPartiallyValidFilter(
-            @Mocked IBackupRestoreConfig backupRestoreConfig,
-            @Capturing INotificationService notificationService)
-            throws ParseException {
-        new Expectations() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                result = "SOME_FAKE_FILE_TYPE_1, SOME_FAKE_FILE_TYPE_2, SST_V2";
-                maxTimes = 2;
-            }
-        };
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 0;
-            }
-        };
-        Path path =
-                Paths.get(
-                        "fakeDataLocation",
-                        "fakeKeyspace",
-                        "fakeColumnFamily",
-                        "fakeBackup",
-                        "fakeData.db");
-        AbstractBackupPath abstractBackupPath = abstractBackupPathProvider.get();
-        abstractBackupPath.parseLocal(path.toFile(), AbstractBackupPath.BackupFileType.META_V2);
-        backupNotificationMgr.notify(abstractBackupPath, UploadStatus.STARTED);
-        new Verifications() {
-            {
-                backupRestoreConfig.getBackupNotifyComponentIncludeList();
-                maxTimes = 2;
-            }
-
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 0;
-            }
-        };
-    }
-
-    @Test
-    public void testNotify(@Capturing INotificationService notificationService) {
-        new Expectations() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-        backupNotificationMgr.notify("some_random", Instant.EPOCH);
-        new Verifications() {
-            {
-                notificationService.notify(anyString, (Map<String, MessageAttributeValue>) any);
-                maxTimes = 1;
-            }
-        };
-    }
-}