diff --git a/aries_cloudagent/config/argparse.py b/aries_cloudagent/config/argparse.py
index 5167831c32..a5b6eabcc9 100644
--- a/aries_cloudagent/config/argparse.py
+++ b/aries_cloudagent/config/argparse.py
@@ -417,6 +417,14 @@ def add_arguments(self, parser: ArgumentParser):
                 "Default: false."
             ),
         )
+        parser.add_argument(
+            "--cred-issue-ack-required",
+            action="store_true",
+            env_var="ACAPY_CRED_ISSUE_ACK_REQUIRED",
+            help=(
+                "An issuer sends please_ack to request a holder to send an ack"
+                "message in response to credential issuance"),
+        )
 
     def get_settings(self, args: Namespace) -> dict:
         """Extract debug settings."""
@@ -468,6 +476,8 @@ def get_settings(self, args: Namespace) -> dict:
             settings["debug.auto_accept_requests"] = True
         if args.auto_respond_messages:
             settings["debug.auto_respond_messages"] = True
+        if args.cred_issue_ack_required:
+            settings["debug.cred_issue_ack_required"] = True
         return settings
 
 
diff --git a/aries_cloudagent/messaging/agent_message.py b/aries_cloudagent/messaging/agent_message.py
index 08e6ec94e9..72428b87e5 100644
--- a/aries_cloudagent/messaging/agent_message.py
+++ b/aries_cloudagent/messaging/agent_message.py
@@ -3,7 +3,7 @@
 import uuid
 from collections import OrderedDict
 from re import sub
-from typing import Mapping, Optional, Text, Union
+from typing import Mapping, Optional, Text, Union, Sequence
 
 from marshmallow import (
     EXCLUDE,
@@ -23,6 +23,7 @@
 from .decorators.service_decorator import ServiceDecorator
 from .decorators.signature_decorator import SignatureDecorator  # TODO deprecated
 from .decorators.thread_decorator import ThreadDecorator
+from .decorators.please_ack_decorator import PleaseAckDecorator
 from .decorators.trace_decorator import (
     TRACE_LOG_TARGET,
     TRACE_MESSAGE_TARGET,
@@ -283,6 +284,32 @@ def _service(self, val: Union[ServiceDecorator, dict]):
         else:
             self._decorators["service"] = val
 
+    @property
+    def _please_ack(self) -> PleaseAckDecorator:
+        """Accessor for the message's please_ack decorator.
+
+        Returns:
+            The PleaseAckDecorator for this message
+        """
+        return self._decorators.get("please_ack")
+
+    @_please_ack.setter
+    def _please_ack(self, val: Union[PleaseAckDecorator, dict]):
+        """Setter for the message's please_ack decorator.
+
+        Args:
+            val: PleaseAckDecorator or dict to set as the please_ack
+        """
+        if val is None:
+            self._decorators.pop("please_ack", None)
+        else:
+            self._decorators["please_ack"] = val
+
+    def assign_please_ack(self, on: Sequence[str]):
+        """Assign a please_ack decorator with specified options
+        """
+        self._please_ack = PleaseAckDecorator(on=on)
+
     @property
     def _thread(self) -> ThreadDecorator:
         """Accessor for the message's thread decorator.
diff --git a/aries_cloudagent/messaging/decorators/default.py b/aries_cloudagent/messaging/decorators/default.py
index bc8c314466..e97555d8d6 100644
--- a/aries_cloudagent/messaging/decorators/default.py
+++ b/aries_cloudagent/messaging/decorators/default.py
@@ -9,6 +9,7 @@
 from .timing_decorator import TimingDecorator
 from .transport_decorator import TransportDecorator
 from .service_decorator import ServiceDecorator
+from .please_ack_decorator import PleaseAckDecorator
 
 DEFAULT_MODELS = {
     "l10n": LocalizationDecorator,
@@ -18,6 +19,7 @@
     "timing": TimingDecorator,
     "transport": TransportDecorator,
     "service": ServiceDecorator,
+    "please_ack": PleaseAckDecorator,
 }
 
 
diff --git a/aries_cloudagent/messaging/decorators/please_ack_decorator.py b/aries_cloudagent/messaging/decorators/please_ack_decorator.py
index 578e2613b8..d5fdb969a3 100644
--- a/aries_cloudagent/messaging/decorators/please_ack_decorator.py
+++ b/aries_cloudagent/messaging/decorators/please_ack_decorator.py
@@ -5,7 +5,6 @@
 from marshmallow import EXCLUDE, fields
 
 from ..models.base import BaseModel, BaseModelSchema
-from ..valid import UUID4_EXAMPLE
 
 
 class PleaseAckDecorator(BaseModel):
@@ -18,8 +17,7 @@ class Meta:
 
     def __init__(
         self,
-        message_id: str = None,
-        on: Sequence[str] = None,
+        on: Sequence[str]
     ):
         """Initialize a PleaseAckDecorator instance.
 
@@ -29,8 +27,7 @@ def __init__(
 
         """
         super().__init__()
-        self.message_id = message_id
-        self.on = list(on) if on else None
+        self.on = list(on)
 
 
 class PleaseAckDecoratorSchema(BaseModelSchema):
@@ -42,14 +39,9 @@ class Meta:
         model_class = PleaseAckDecorator
         unknown = EXCLUDE
 
-    message_id = fields.Str(
-        required=False,
-        allow_none=False,
-        metadata={"description": "Message identifier", "example": UUID4_EXAMPLE},
-    )
     on = fields.List(
         fields.Str(metadata={"example": "OUTCOME"}),
-        required=False,
+        required=True,
         metadata={
             "description": "List of tokens describing circumstances for acknowledgement"
         },
diff --git a/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_issue_handler.py b/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_issue_handler.py
index f3c5704252..9e13b7c40a 100644
--- a/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_issue_handler.py
+++ b/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_issue_handler.py
@@ -89,7 +89,11 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
                         )
                     )
 
-            cred_ack_message = await cred_manager.send_cred_ack(cred_ex_record)
+            if cred_ex_record.ack_required:
+                cred_ack_message = await cred_manager.send_cred_ack(cred_ex_record)
+            else:
+                cred_ack_message = await cred_manager.transit_to_done(cred_ex_record)
+
 
             trace_event(
                 context.settings,
diff --git a/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py b/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py
index ecf0d6fb15..8ac2332eeb 100644
--- a/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py
+++ b/aries_cloudagent/protocols/issue_credential/v2_0/handlers/cred_request_handler.py
@@ -14,6 +14,7 @@
 from ..manager import V20CredManager, V20CredManagerError
 from ..messages.cred_problem_report import ProblemReportReason
 from ..messages.cred_request import V20CredRequest
+from ..models.cred_ex_record import V20CredExRecord
 
 
 class V20CredRequestHandler(BaseHandler):
@@ -97,6 +98,10 @@ async def handle(self, context: RequestContext, responder: BaseResponder):
                     )
                 )
 
+            if cred_ex_record.state == V20CredExRecord.STATE_DONE and cred_ex_record.auto_remove:
+                self._logger.debug("delete cred_ex_record")
+                await cred_manager.delete_cred_ex_record(cred_ex_record.cred_ex_id)
+
             trace_event(
                 context.settings,
                 cred_issue_message,
diff --git a/aries_cloudagent/protocols/issue_credential/v2_0/manager.py b/aries_cloudagent/protocols/issue_credential/v2_0/manager.py
index fe737c27f6..ad8f56e49a 100644
--- a/aries_cloudagent/protocols/issue_credential/v2_0/manager.py
+++ b/aries_cloudagent/protocols/issue_credential/v2_0/manager.py
@@ -515,8 +515,15 @@ async def issue_credential(
             credentials_attach=[attach for (_, attach) in issue_formats],
         )
 
-        cred_ex_record.state = V20CredExRecord.STATE_ISSUED
         cred_ex_record.cred_issue = cred_issue_message
+
+        if self._profile.settings.get("debug.cred_issue_ack_required"):
+            # request the holder to answer with an ack message
+            cred_ex_record.state = V20CredExRecord.STATE_ISSUED
+            cred_issue_message.assign_please_ack(["OUTCOME"])
+        else:
+            cred_ex_record.state = V20CredExRecord.STATE_DONE
+
         async with self._profile.session() as session:
             # FIXME - re-fetch record to check state, apply transactional update
             await cred_ex_record.save(session, reason="v2.0 issue credential")
@@ -550,6 +557,11 @@ async def receive_credential(
                 role=V20CredExRecord.ROLE_HOLDER,
             )
 
+        please_ack = cred_issue_message._please_ack
+
+        if please_ack is not None and 'OUTCOME' in please_ack.on:
+            cred_ex_record.ack_required = True
+
         cred_request_message = cred_ex_record.cred_request
         req_formats = [
             V20CredFormat.Format.get(fmt.format)
@@ -667,6 +679,36 @@ async def send_cred_ack(
 
         return cred_ex_record, cred_ack_message
 
+
+    async def transit_to_done(
+        self,
+        cred_ex_record: V20CredExRecord,
+    ):
+        """Transition of the protocol instance to STATE_DONE
+
+        Delete cred ex record if set to auto-remove.
+
+        Returns:
+            cred ex record
+        """
+
+        # FIXME - most of the code are copy-pasted from the send_cred_ack()
+        cred_ex_record.state = V20CredExRecord.STATE_DONE
+        try:
+            async with self._profile.session() as session:
+                await cred_ex_record.save(session, reason="store credential v2.0")
+
+            if cred_ex_record.auto_remove:
+                await self.delete_cred_ex_record(cred_ex_record.cred_ex_id)
+
+        except StorageError:
+            LOGGER.exception(
+                "Error transition to done"
+            )
+
+        return cred_ex_record
+
+
     async def receive_credential_ack(
         self, cred_ack_message: V20CredAck, connection_id: Optional[str]
     ) -> V20CredExRecord:
diff --git a/aries_cloudagent/protocols/issue_credential/v2_0/models/cred_ex_record.py b/aries_cloudagent/protocols/issue_credential/v2_0/models/cred_ex_record.py
index 1eece14bac..e28266c246 100644
--- a/aries_cloudagent/protocols/issue_credential/v2_0/models/cred_ex_record.py
+++ b/aries_cloudagent/protocols/issue_credential/v2_0/models/cred_ex_record.py
@@ -69,6 +69,7 @@ def __init__(
         auto_offer: bool = False,
         auto_issue: bool = False,
         auto_remove: bool = True,
+        ack_required: bool = False,
         error_msg: str = None,
         trace: bool = False,  # backward compat: BaseRecord.from_storage()
         cred_id_stored: str = None,  # backward compat: BaseRecord.from_storage()
@@ -93,6 +94,7 @@ def __init__(
         self.auto_offer = auto_offer
         self.auto_issue = auto_issue
         self.auto_remove = auto_remove
+        self.ack_required = ack_required
         self.error_msg = error_msg
 
     @property
@@ -222,6 +224,7 @@ def record_value(self) -> Mapping:
                     "auto_offer",
                     "auto_issue",
                     "auto_remove",
+                    "ack_required",
                     "error_msg",
                     "trace",
                 )
@@ -425,6 +428,16 @@ class Meta:
             "example": False,
         },
     )
+    ack_required = fields.Bool(
+        required=False,
+        dump_default=False,
+        metadata={
+            "description": (
+                "Issuer requests to send ack in response to credentials"
+            ),
+            "example": False,
+        },
+    )
     error_msg = fields.Str(
         required=False,
         metadata={"description": "Error message", "example": "The front fell off"},
diff --git a/aries_cloudagent/protocols/issue_credential/v2_0/routes.py b/aries_cloudagent/protocols/issue_credential/v2_0/routes.py
index 721128b4ae..a891e5de86 100644
--- a/aries_cloudagent/protocols/issue_credential/v2_0/routes.py
+++ b/aries_cloudagent/protocols/issue_credential/v2_0/routes.py
@@ -1559,11 +1559,15 @@ async def credential_exchange_store(request: web.BaseRequest):
         # fetch these early, before potential removal
         details = await _get_attached_credentials(profile, cred_ex_record)
 
-        # the record may be auto-removed here
-        (
-            cred_ex_record,
-            cred_ack_message,
-        ) = await cred_manager.send_cred_ack(cred_ex_record)
+        if cred_ex_record.ack_required:
+            # the record may be auto-removed here
+            (
+                cred_ex_record,
+                cred_ack_message,
+            ) = await cred_manager.send_cred_ack(cred_ex_record)
+        else:
+            cred_ex_record = await cred_manager.transit_to_done(cred_ex_record)
+            cred_ack_message = None
 
         result = _format_result_with_details(cred_ex_record, details)
 
diff --git a/demo/runners/support/agent.py b/demo/runners/support/agent.py
index 26563fe6f5..cbbc7f6733 100644
--- a/demo/runners/support/agent.py
+++ b/demo/runners/support/agent.py
@@ -332,6 +332,8 @@ def get_agent_args(self):
             ("--endpoint", self.endpoint),
             ("--label", self.label),
             "--auto-ping-connection",
+            # "--cred-issue-ack-required", # defines whether an issuer requires
+                                           # an explicit ack message from a holder
             "--auto-respond-messages",
             ("--inbound-transport", "http", "0.0.0.0", str(self.http_port)),
             ("--outbound-transport", "http"),