[Issue 3926][functions] Add delete counter in python functions state_…

…context (apache#3930)

* add delete_counter in python functions state_context

* Update contextimpl.py

Matched the name of the stub method to the one in ContextImpl

pulsar-client-cpp/python/pulsar/functions/context.py

@@ -154,6 +154,11 @@ def get_counter(self, key):
     """get the counter of a given key in the managed state"""
     pass
 
+  @abstractmethod
+  def del_counter(self, key):
+    """delete the counter of a given key in the managed state"""
+    pass
+
   @abstractmethod
   def put_state(self, key, value):
     """update the value of a given key in the managed state"""

pulsar-functions/instance/src/main/python/contextimpl.py

@@ -215,6 +215,9 @@ def incr_counter(self, key, amount):
   def get_counter(self, key):
     return self.state_context.get_amount(key)
 
+  def del_counter(self, key):
+    return self.state_context.delete(key)
+
   def put_state(self, key, value):
     return self.state_context.put(key, value)
 

pulsar-functions/instance/src/main/python/state_context.py

@@ -106,9 +106,9 @@ class BKManagedStateContext(StateContext):
 
     def __init__(self, state_storage_serviceurl, table_ns, table_name):
         client_settings = StorageClientSettings(
-          service_uri=state_storage_serviceurl)
+            service_uri=state_storage_serviceurl)
         admin_client = admin.client.Client(
-          storage_client_settings=client_settings)
+            storage_client_settings=client_settings)
         # create namespace and table if needed
         ns = admin_client.namespace(table_ns)
         try:
@@ -154,3 +154,6 @@ def get_value(self, key):
 
     def put(self, key, value):
         return self.__table__.put_str(key, value)
+
+    def delete_key(self, key):
+        return self.__table__.delete_str(key)