class RetentionLeaseHelper {

    private static final Logger LOGGER = LogManager.getLogger(RetentionLeaseHelper.class);

    private static String retentionLeaseSource(String subscriberClusterName) {
        return "logical_replication:" + subscriberClusterName;
    }

    private static String retentionLeaseIdForShard(String subscriberClusterName, ShardId shardId) {
        var retentionLeaseSource = retentionLeaseSource(subscriberClusterName);
        return retentionLeaseSource + ":" + shardId;
    }


    public static void addRetentionLease(ShardId shardId,
                                         long seqNo,
                                         String subscriberClusterName,
                                         Client client,
                                         ActionListener<RetentionLeaseActions.Response> listener) {<FILL_FUNCTION_BODY>}

    public static void renewRetentionLease(ShardId shardId,
                                           long seqNo,
                                           String subscriberClusterName,
                                           Client client,
                                           ActionListener<RetentionLeaseActions.Response> listener) {
        var retentionLeaseId = retentionLeaseIdForShard(subscriberClusterName, shardId);
        var request = new RetentionLeaseActions.AddOrRenewRequest(
            shardId,
            retentionLeaseId,
            seqNo,
            retentionLeaseSource(subscriberClusterName)
        );
        client.execute(RetentionLeaseActions.Renew.INSTANCE, request).whenComplete(listener);
    }

    public static void attemptRetentionLeaseRemoval(ShardId shardId,
                                                    String subscriberClusterName,
                                                    Client client,
                                                    ActionListener<RetentionLeaseActions.Response> listener) {
        var retentionLeaseId = retentionLeaseIdForShard(subscriberClusterName, shardId);
        var request = new RetentionLeaseActions.RemoveRequest(shardId, retentionLeaseId);
        client.execute(RetentionLeaseActions.Remove.INSTANCE, request)
            .whenComplete((response, err) -> {
                if (err == null) {
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("Removed retention lease with id - {}", retentionLeaseId);
                    }
                    listener.onResponse(response);
                } else {
                    var e = Exceptions.toException(SQLExceptions.unwrap(err));
                    if (LOGGER.isDebugEnabled()) {
                        LOGGER.debug("Exception in removing retention lease", e);
                    }
                    listener.onFailure(e);
                }
            });
    }
}

        var retentionLeaseId = retentionLeaseIdForShard(subscriberClusterName, shardId);
        var request = new RetentionLeaseActions.AddOrRenewRequest(
            shardId,
            retentionLeaseId,
            seqNo,
            retentionLeaseSource(subscriberClusterName)
        );
        client.execute(RetentionLeaseActions.Add.INSTANCE, request)
            .whenComplete((response, err) -> {
                if (err == null) {
                    listener.onResponse(response);
                } else {
                    var t = SQLExceptions.unwrap(err);
                    if (t instanceof RetentionLeaseAlreadyExistsException) {
                        if (LOGGER.isDebugEnabled()) {
                            LOGGER.debug(
                                "Renew retention lease as it already exists {} with {}",
                                retentionLeaseId,
                                seqNo
                            );
                        }
                        // Only one retention lease should exists for the follower shard
                        // Ideally, this should have got cleaned-up
                        renewRetentionLease(shardId, seqNo, subscriberClusterName, client, listener);
                    } else {
                        listener.onFailure(Exceptions.toException(t));
                    }
                }
            });
    
<no_super_class>