Read cluster shard count from DB (#3788)

Author: yux0

common/cluster/metadata.go

@@ -104,7 +104,7 @@ type (
 		InitialFailoverVersion int64 `yaml:"initialFailoverVersion"`
 		// Address indicate the remote service address(Host:Port). Host can be DNS name.
 		RPCAddress string `yaml:"rpcAddress"`
-		ShardCount int32
+		ShardCount int32  `yaml:"-"` // Ignore this field when loading config.
 		// private field to track cluster information updates
 		version int64
 	}
@@ -401,6 +401,7 @@ func (m *metadataImpl) RegisterMetadataChangeCallback(callbackId any, cb Callbac
 			Enabled:                clusterInfo.Enabled,
 			InitialFailoverVersion: clusterInfo.InitialFailoverVersion,
 			RPCAddress:             clusterInfo.RPCAddress,
+			ShardCount:             clusterInfo.ShardCount,
 			version:                clusterInfo.version,
 		}
 	}
@@ -454,6 +455,7 @@ func (m *metadataImpl) refreshClusterMetadata(ctx context.Context) error {
 				Enabled:                newClusterInfo.Enabled,
 				InitialFailoverVersion: newClusterInfo.InitialFailoverVersion,
 				RPCAddress:             newClusterInfo.RPCAddress,
+				ShardCount:             newClusterInfo.ShardCount,
 				version:                newClusterInfo.version,
 			}
 		} else if newClusterInfo.version > oldClusterInfo.version {
@@ -468,12 +470,14 @@ func (m *metadataImpl) refreshClusterMetadata(ctx context.Context) error {
 				Enabled:                oldClusterInfo.Enabled,
 				InitialFailoverVersion: oldClusterInfo.InitialFailoverVersion,
 				RPCAddress:             oldClusterInfo.RPCAddress,
+				ShardCount:             oldClusterInfo.ShardCount,
 				version:                oldClusterInfo.version,
 			}
 			newEntries[clusterName] = &ClusterInformation{
 				Enabled:                newClusterInfo.Enabled,
 				InitialFailoverVersion: newClusterInfo.InitialFailoverVersion,
 				RPCAddress:             newClusterInfo.RPCAddress,
+				ShardCount:             newClusterInfo.ShardCount,
 				version:                newClusterInfo.version,
 			}
 		}
@@ -577,6 +581,7 @@ func (m *metadataImpl) listAllClusterMetadataFromDB(
 			Enabled:                getClusterResp.GetIsConnectionEnabled(),
 			InitialFailoverVersion: getClusterResp.GetInitialFailoverVersion(),
 			RPCAddress:             getClusterResp.GetClusterAddress(),
+			ShardCount:             getClusterResp.GetHistoryShardCount(),
 			version:                getClusterResp.Version,
 		}
 	}

common/cluster/metadata_test.go

@@ -83,12 +83,14 @@ func (s *metadataSuite) SetupTest() {
 			Enabled:                true,
 			InitialFailoverVersion: int64(1),
 			RPCAddress:             uuid.New(),
+			ShardCount:             1,
 			version:                1,
 		},
 		s.secondClusterName: {
 			Enabled:                true,
 			InitialFailoverVersion: int64(4),
 			RPCAddress:             uuid.New(),
+			ShardCount:             2,
 			version:                1,
 		},
 	}
@@ -174,6 +176,7 @@ func (s *metadataSuite) Test_RefreshClusterMetadata_Success() {
 						ClusterName:            s.clusterName,
 						IsConnectionEnabled:    true,
 						InitialFailoverVersion: 1,
+						HistoryShardCount:      1,
 						ClusterAddress:         uuid.New(),
 					},
 					Version: 1,
@@ -183,6 +186,7 @@ func (s *metadataSuite) Test_RefreshClusterMetadata_Success() {
 						ClusterName:            id,
 						IsConnectionEnabled:    true,
 						InitialFailoverVersion: 2,
+						HistoryShardCount:      2,
 						ClusterAddress:         uuid.New(),
 					},
 					Version: 2,
@@ -207,6 +211,7 @@ func (s *metadataSuite) Test_ListAllClusterMetadataFromDB_Success() {
 						ClusterName:            s.clusterName,
 						IsConnectionEnabled:    true,
 						InitialFailoverVersion: 1,
+						HistoryShardCount:      1,
 						ClusterAddress:         uuid.New(),
 					},
 					Version: 1,
@@ -225,6 +230,7 @@ func (s *metadataSuite) Test_ListAllClusterMetadataFromDB_Success() {
 						ClusterName:            newClusterName,
 						IsConnectionEnabled:    true,
 						InitialFailoverVersion: 2,
+						HistoryShardCount:      2,
 						ClusterAddress:         uuid.New(),
 					},
 					Version: 2,

temporal/fx.go

@@ -767,10 +767,16 @@ func loadClusterInformationFromStore(ctx context.Context, config *config.Config,
 			return err
 		}
 		metadata := item.(*persistence.GetClusterMetadataResponse)
+		shardCount := metadata.HistoryShardCount
+		if shardCount == 0 {
+			// This is to add backward compatibility to the config based cluster connection.
+			shardCount = config.Persistence.NumHistoryShards
+		}
 		newMetadata := cluster.ClusterInformation{
 			Enabled:                metadata.IsConnectionEnabled,
 			InitialFailoverVersion: metadata.InitialFailoverVersion,
 			RPCAddress:             metadata.ClusterAddress,
+			ShardCount:             shardCount,
 		}
 		if staticClusterMetadata, ok := config.ClusterMetadata.ClusterInformation[metadata.ClusterName]; ok {
 			if metadata.ClusterName != config.ClusterMetadata.CurrentClusterName {