state: Adds io.Closer to interface

state/aerospike/aerospike.go

@@ -246,6 +246,10 @@ func (aspike *Aerospike) Delete(ctx context.Context, req *state.DeleteRequest) e
 	return nil
 }
 
+func (aspike *Aerospike) Close() error {
+	return nil
+}
+
 func parseHosts(hostsMeta string) ([]*as.Host, error) {
 	hostPorts := []*as.Host{}
 	for _, hostPort := range strings.Split(hostsMeta, ",") {

state/alicloud/tablestore/tablestore.go

@@ -234,3 +234,7 @@ func (s *AliCloudTableStore) GetComponentMetadata() (metadataInfo metadata.Metad
 	metadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, metadata.StateStoreType)
 	return
 }
+
+func (s *AliCloudTableStore) Close() error {
+	return nil
+}

state/aws/dynamodb/dynamodb.go

@@ -267,6 +267,10 @@ func (d *StateStore) GetComponentMetadata() (metadataInfo metadata.MetadataMap)
 	return
 }
 
+func (d *StateStore) Close() error {
+	return nil
+}
+
 func (d *StateStore) getDynamoDBMetadata(meta state.Metadata) (*dynamoDBMetadata, error) {
 	var m dynamoDBMetadata
 	err := kitmd.DecodeMetadata(meta.Properties, &m)

state/azure/blobstorage/internal/blobstorage.go

@@ -119,6 +119,10 @@ func (r *StateStore) readFile(ctx context.Context, req *state.GetRequest) (*stat
 	}, nil
 }
 
+func (r *StateStore) Close() error {
+	return nil
+}
+
 func (r *StateStore) writeFile(ctx context.Context, req *state.SetRequest) error {
 	modifiedAccessConditions := blob.ModifiedAccessConditions{}
 

state/azure/cosmosdb/cosmosdb.go

@@ -623,6 +623,10 @@ func (c *StateStore) Ping(ctx context.Context) error {
 	return nil
 }
 
+func (c *StateStore) Close() error {
+	return nil
+}
+
 func createUpsertItem(contentType string, req state.SetRequest, partitionKey string) (CosmosItem, error) {
 	byteArray, isBinary := req.Value.([]byte)
 	if len(byteArray) == 0 {

state/azure/tablestorage/tablestorage.go

@@ -211,6 +211,10 @@ func (r *StateStore) GetComponentMetadata() (metadataInfo mdutils.MetadataMap) {
 	return
 }
 
+func (r *StateStore) Close() error {
+	return nil
+}
+
 func NewAzureTablesStateStore(logger logger.Logger) state.Store {
 	s := &StateStore{
 		json:     jsoniter.ConfigFastest,

state/bulk_test.go

@@ -164,6 +164,10 @@ func (s *storeBulk) Set(ctx context.Context, req *SetRequest) error {
 	return nil
 }
 
+func (s *storeBulk) Close() error {
+	return nil
+}
+
 func (s *storeBulk) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 	return
 }

state/hashicorp/consul/consul.go

@@ -163,3 +163,7 @@ func (c *Consul) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 	metadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, metadata.StateStoreType)
 	return
 }
+
+func (c *Consul) Close() error {
+	return nil
+}

state/hazelcast/hazelcast.go

@@ -165,3 +165,7 @@ func (store *Hazelcast) GetComponentMetadata() (metadataInfo metadata.MetadataMa
 	metadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, metadata.StateStoreType)
 	return
 }
+
+func (store *Hazelcast) Close() error {
+	return nil
+}

state/jetstream/jetstream.go

@@ -16,7 +16,6 @@ package jetstream
 import (
 	"context"
 	"errors"
-	"io"
 	"reflect"
 	"strings"
 	"sync/atomic"
@@ -186,5 +185,3 @@ func (js *StateStore) Close() error {
 	}
 	return nil
 }
-
-var _ io.Closer = (*StateStore)(nil)

state/mongodb/mongodb.go

@@ -693,10 +693,12 @@ func (m *MongoDB) GetComponentMetadata() (metadataInfo metadata.MetadataMap) {
 }
 
 // Close connection to the database.
-func (m *MongoDB) Close(ctx context.Context) (err error) {
+func (m *MongoDB) Close() error {
 	if m.client == nil {
 		return nil
 	}
 
+	ctx, cancel := context.WithTimeout(context.Background(), time.Second*5)
+	defer cancel()
 	return m.client.Disconnect(ctx)
 }

state/oci/objectstorage/objectstorage.go

@@ -167,6 +167,10 @@ func (r *StateStore) Ping(ctx context.Context) error {
 	return r.pingBucket(ctx)
 }
 
+func (r *StateStore) Close() error {
+	return nil
+}
+
 func NewOCIObjectStorageStore(logger logger.Logger) state.Store {
 	s := &StateStore{
 		json: jsoniter.ConfigFastest,

state/oracledatabase/oracledatabase_integration_test.go

@@ -21,7 +21,6 @@ import (
 	"encoding/json"
 	"errors"
 	"fmt"
-	"io"
 	"os"
 	"testing"
 	"time"
@@ -69,7 +68,7 @@ func TestOracleDatabaseIntegration(t *testing.T) {
 
 	ods := NewOracleDatabaseStateStore(logger.NewLogger("test"))
 	t.Cleanup(func() {
-		defer ods.(io.Closer).Close()
+		defer ods.Close()
 	})
 
 	if initerror := ods.Init(context.Background(), metadata); initerror != nil {
@@ -672,7 +671,7 @@ func testInitConfiguration(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			p := NewOracleDatabaseStateStore(logger)
-			defer p.(io.Closer).Close()
+			defer p.Close()
 
 			metadata := state.Metadata{
 				Base: metadata.Base{Properties: tt.props},

state/rethinkdb/rethinkdb_test.go

@@ -17,7 +17,6 @@ import (
 	"context"
 	"encoding/json"
 	"fmt"
-	"io"
 	"os"
 	"strconv"
 	"testing"
@@ -166,9 +165,7 @@ func TestRethinkDBStateStoreRongRun(t *testing.T) {
 	if err := db.Init(context.Background(), m); err != nil {
 		t.Fatalf("error initializing db: %v", err)
 	}
-	closer, ok := db.(io.Closer)
-	assert.True(t, ok)
-	defer require.NoError(t, closer.Close())
+	defer require.NoError(t, db.Close())
 
 	for i := 0; i < 1000; i++ {
 		testBulk(t, db, i)

state/sqlite/sqlite_integration_test.go

@@ -18,7 +18,6 @@ import (
 	"database/sql"
 	"encoding/json"
 	"fmt"
-	"io"
 	"os"
 	"sort"
 	"testing"
@@ -56,7 +55,7 @@ func TestSqliteIntegration(t *testing.T) {
 
 	s := NewSQLiteStateStore(logger.NewLogger("test"))
 	t.Cleanup(func() {
-		defer s.(io.Closer).Close()
+		defer s.Close()
 	})
 
 	if initerror := s.Init(context.Background(), metadata); initerror != nil {
@@ -693,7 +692,7 @@ func testInitConfiguration(t *testing.T) {
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			p := NewSQLiteStateStore(logger)
-			defer p.(io.Closer).Close()
+			defer p.Close()
 
 			metadata := state.Metadata{
 				Base: metadata.Base{

state/store.go

@@ -16,6 +16,7 @@ package state
 import (
 	"context"
 	"errors"
+	"io"
 
 	"github.com/dapr/components-contrib/health"
 	"github.com/dapr/components-contrib/metadata"
@@ -39,6 +40,7 @@ type BaseStore interface {
 	Delete(ctx context.Context, req *DeleteRequest) error
 	Get(ctx context.Context, req *GetRequest) (*GetResponse, error)
 	Set(ctx context.Context, req *SetRequest) error
+	io.Closer
 }
 
 // TransactionalStore is an interface for initialization and support multiple transactional requests.

state/zookeeper/zk.go

@@ -309,3 +309,7 @@ func (s *StateStore) GetComponentMetadata() (metadataInfo metadata.MetadataMap)
 	metadata.GetMetadataInfoFromStructType(reflect.TypeOf(metadataStruct), &metadataInfo, metadata.StateStoreType)
 	return
 }
+
+func (s *StateStore) Close() error {
+	return nil
+}