fix(coordinator): panic when BadVersion in coordinator (#512)

## Motivation

The coordinator will stop all the operations when it stuck in the
metadata `bad version` case. this PR is trying to trigger panic when bad
version to help auto-restart. the bad version might caused by many
cases:


1. operator/maintainer update the status config.
2. another coordinator updated metadata before shutdown.


> the coordinator is using lock to ensure all the metadata operation is
invoke sequentially. which means we don't have race conditions in the
process.

## Modification

- panic directly when met `BadVersion`

## Alternatives

- Update to the latest version when update metadata get `BadVersion`.
but we have to change some logics to ensure get version before building
the new status to avoid unexpected override.

coordinator/impl/metadata_configmap.go

@@ -103,13 +103,13 @@ func (m *metadataProviderConfigMap) Store(status *model.ClusterStatus, expectedV
 	}
 
 	if version != expectedVersion {
-		return version, ErrMetadataBadVersion
+		panic(ErrMetadataBadVersion)
 	}
 
 	data := configMap(m.name, status, expectedVersion)
 	cm, err := K8SConfigMaps(m.kubernetes).Upsert(m.namespace, m.name, data)
 	if k8serrors.IsConflict(err) {
-		return version, ErrMetadataBadVersion
+		panic(ErrMetadataBadVersion)
 	}
 	version = Version(cm.ResourceVersion)
 	m.metadataSize.Store(int64(len(data.Data["status"])))

coordinator/impl/metadata_file.go

@@ -101,7 +101,7 @@ func (m *metadataProviderFile) Store(cs *model.ClusterStatus, expectedVersion Ve
 	}
 
 	if expectedVersion != existingVersion {
-		return MetadataNotExists, ErrMetadataBadVersion
+		panic(ErrMetadataBadVersion)
 	}
 
 	newVersion = incrVersion(existingVersion)

coordinator/impl/metadata_memory.go

@@ -52,7 +52,7 @@ func (m *metadataProviderMemory) Store(cs *model.ClusterStatus, expectedVersion
 	defer m.Unlock()
 
 	if expectedVersion != m.version {
-		return MetadataNotExists, ErrMetadataBadVersion
+		panic(ErrMetadataBadVersion)
 	}
 
 	m.cs = cs.Clone()

coordinator/impl/metadata_test.go

@@ -59,13 +59,14 @@ func TestMetadataProvider(t *testing.T) {
 			assert.Equal(t, MetadataNotExists, version)
 			assert.Nil(t, res)
 
-			newVersion, err := m.Store(&model.ClusterStatus{
-				Namespaces: map[string]model.NamespaceStatus{},
-			}, "")
-			assert.ErrorIs(t, err, ErrMetadataBadVersion)
-			assert.Equal(t, MetadataNotExists, newVersion)
+			assert.PanicsWithError(t, ErrMetadataBadVersion.Error(), func() {
+				_, err := m.Store(&model.ClusterStatus{
+					Namespaces: map[string]model.NamespaceStatus{},
+				}, "")
+				assert.NoError(t, err)
+			})
 
-			newVersion, err = m.Store(&model.ClusterStatus{
+			newVersion, err := m.Store(&model.ClusterStatus{
 				Namespaces: map[string]model.NamespaceStatus{},
 			}, MetadataNotExists)
 			assert.NoError(t, err)