632
633
// TestConcurrentUpdateNamespace test
635
>
id := uuid.NewString()
636
>
name := "concurrent-update-namespace-test-name"
637
>
state := enumspb.NAMESPACE_STATE_REGISTERED
638
>
description := "update-namespace-test-description"
639
>
owner := "update-namespace-test-owner"
640
>
data := map[string]string{"k1": "v1"}
641
>
retention := int32(10)
642
>
historyArchivalState := enumspb.ARCHIVAL_STATE_ENABLED
643
>
historyArchivalURI := "test://history/uri"
644
>
visibilityArchivalState := enumspb.ARCHIVAL_STATE_ENABLED
645
>
visibilityArchivalURI := "test://visibility/uri"
646
>
badBinaries := &namespacepb.BadBinaries{Binaries: map[string]*namespacepb.BadBinaryInfo{}}
647
>
648
>
clusterActive := "some random active cluster name"
649
>
clusterStandby := "some random standby cluster name"
650
>
configVersion := int64(10)
651
>
failoverVersion := int64(59)
652
>
isGlobalNamespace := true
653
>
clusters := []string{clusterActive, clusterStandby}
654
>
655
>
resp1, err1 := m.CreateNamespace(
656
>
&persistencespb.NamespaceInfo{
657
>
Id: id,
658
>
Name: name,
659
>
State: state,
660
>
Description: description,
661
>
Owner: owner,
662
>
Data: data,
663
>
},
664
>
&persistencespb.NamespaceConfig{
665
>
Retention: timestamp.DurationFromDays(retention),
666
>
HistoryArchivalState: historyArchivalState,
667
>
HistoryArchivalUri: historyArchivalURI,
668
>
VisibilityArchivalState: visibilityArchivalState,
669
>
VisibilityArchivalUri: visibilityArchivalURI,
670
>
BadBinaries: badBinaries,
671
>
},
672
>
&persistencespb.NamespaceReplicationConfig{
673
>
ActiveClusterName: clusterActive,
674
>
Clusters: clusters,
675
>
},
676
>
isGlobalNamespace,
677
>
configVersion,
678
>
failoverVersion,
679
>
)
680
>
m.NoError(err1)
681
>
m.EqualValues(id, resp1.ID)
682
>
683
>
resp2, err2 := m.GetNamespace(id, "")
684
>
m.NoError(err2)
685
>
m.ProtoEqual(badBinaries, resp2.Namespace.Config.BadBinaries)
686
>
metadata, err := m.MetadataManager.GetMetadata(m.ctx)
687
>
m.NoError(err)
688
>
notificationVersion := metadata.NotificationVersion
689
>
690
>
testBinaries := &namespacepb.BadBinaries{
691
>
Binaries: map[string]*namespacepb.BadBinaryInfo{
692
>
"abc": {
693
>
Reason: "test-reason",
694
>
Operator: "test-operator",
695
>
CreateTime: timestamppb.New(time.Date(2020, 8, 22, 0, 0, 0, 0, time.UTC)),
696
>
},
697
>
},
698
>
}
699
>
concurrency := 16
700
>
successCount := int32(0)
701
>
var wg sync.WaitGroup
702
>
for i := 1; i <= concurrency; i++ {
703
>
newValue := fmt.Sprintf("v-%v", i)
704
>
wg.Add(1)
705
>
go func(updatedData map[string]string) {
706
>
err3 := m.UpdateNamespace(
707
>
&persistencespb.NamespaceInfo{
708
>
Id: resp2.Namespace.Info.Id,
709
>
Name: resp2.Namespace.Info.Name,
710
>
State: resp2.Namespace.Info.State,
711
>
Description: resp2.Namespace.Info.Description,
712
>
Owner: resp2.Namespace.Info.Owner,
713
>
Data: updatedData,
714
>
},
715
>
&persistencespb.NamespaceConfig{
716
>
Retention: resp2.Namespace.Config.Retention,
717
>
HistoryArchivalState: resp2.Namespace.Config.HistoryArchivalState,
718
>
HistoryArchivalUri: resp2.Namespace.Config.HistoryArchivalUri,
719
>
VisibilityArchivalState: resp2.Namespace.Config.VisibilityArchivalState,
720
>
VisibilityArchivalUri: resp2.Namespace.Config.VisibilityArchivalUri,
721
>
BadBinaries: testBinaries,
722
>
},
723
>
&persistencespb.NamespaceReplicationConfig{
724
>
ActiveClusterName: resp2.Namespace.ReplicationConfig.ActiveClusterName,
725
>
Clusters: resp2.Namespace.ReplicationConfig.Clusters,
726
>
},
727
>
resp2.Namespace.ConfigVersion,
728
>
resp2.Namespace.FailoverVersion,
729
>
resp2.Namespace.FailoverNotificationVersion,
730
>
time.Time{},
731
>
notificationVersion,
732
>
isGlobalNamespace,
733
>
)
734
>
if err3 == nil {
735
>
atomic.AddInt32(&successCount, 1)
736
>
}
737
>
wg.Done()
738
}(map[string]string{"k0": newValue})
739
}
741
>
m.Equal(int32(1), successCount)
742
>
743
>
resp3, err3 := m.GetNamespace("", name)
744
>
m.NoError(err3)
745
>
m.NotNil(resp3)
746
>
m.EqualValues(id, resp3.Namespace.Info.Id)
747
>
m.Equal(name, resp3.Namespace.Info.Name)
748
>
m.Equal(state, resp3.Namespace.Info.State)
749
>
m.Equal(isGlobalNamespace, resp3.IsGlobalNamespace)
750
>
m.Equal(description, resp3.Namespace.Info.Description)
751
>
m.Equal(owner, resp3.Namespace.Info.Owner)
752
>
753
>
m.EqualValues(time.Duration(retention)*time.Hour*24, resp3.Namespace.Config.Retention.AsDuration())
754
>
m.Equal(historyArchivalState, resp3.Namespace.Config.HistoryArchivalState)
755
>
m.Equal(historyArchivalURI, resp3.Namespace.Config.HistoryArchivalUri)
756
>
m.Equal(visibilityArchivalState, resp3.Namespace.Config.VisibilityArchivalState)
757
>
m.Equal(visibilityArchivalURI, resp3.Namespace.Config.VisibilityArchivalUri)
758
>
m.ProtoEqual(testBinaries, resp3.Namespace.Config.BadBinaries)
759
>
m.Equal(clusterActive, resp3.Namespace.ReplicationConfig.ActiveClusterName)
760
>
m.Equal(len(clusters), len(resp3.Namespace.ReplicationConfig.Clusters))
761
>
for index := range clusters {
762
>
m.Equal(clusters[index], resp3.Namespace.ReplicationConfig.Clusters[index])
763
>
}
764
>
m.Equal(isGlobalNamespace, resp3.IsGlobalNamespace)
765
>
m.Equal(configVersion, resp3.Namespace.ConfigVersion)
766
>
m.Equal(failoverVersion, resp3.Namespace.FailoverVersion)
767
>
768
>
// check namespace data
769
>
ss := strings.Split(resp3.Namespace.Info.Data["k0"], "-")
770
>
m.Equal(2, len(ss))
771
>
vi, err := strconv.Atoi(ss[1])
772
>
m.NoError(err)
773
>
m.Equal(true, vi > 0 && vi <= concurrency)
774
}
775