531
532
// TestConcurrentCreateNamespace test
533
>
func (m *MetadataPersistenceSuiteV2) TestConcurrentCreateNamespace() {
metadata_persistence_v2.go
534
>
id := uuid.NewString()
535
>
536
>
name := "concurrent-create-namespace-test-name"
537
>
state := enumspb.NAMESPACE_STATE_REGISTERED
538
>
description := "concurrent-create-namespace-test-description"
539
>
owner := "create-namespace-test-owner"
540
>
retention := int32(10)
541
>
historyArchivalState := enumspb.ARCHIVAL_STATE_ENABLED
542
>
historyArchivalURI := "test://history/uri"
543
>
visibilityArchivalState := enumspb.ARCHIVAL_STATE_ENABLED
544
>
visibilityArchivalURI := "test://visibility/uri"
545
>
546
>
clusterActive := "some random active cluster name"
547
>
clusterStandby := "some random standby cluster name"
548
>
configVersion := int64(10)
549
>
failoverVersion := int64(59)
550
>
isGlobalNamespace := true
551
>
clusters := []string{clusterActive, clusterStandby}
552
>
553
>
testBinaries := &namespacepb.BadBinaries{
554
>
Binaries: map[string]*namespacepb.BadBinaryInfo{
555
>
"abc": {
556
>
Reason: "test-reason",
557
>
Operator: "test-operator",
558
>
CreateTime: timestamppb.New(time.Date(2020, 8, 22, 0, 0, 0, 0, time.UTC)),
559
>
},
560
>
},
561
>
}
562
>
concurrency := 16
563
>
successCount := int32(0)
564
>
var wg sync.WaitGroup
565
>
for i := 1; i <= concurrency; i++ {
566
>
newValue := fmt.Sprintf("v-%v", i)
567
>
wg.Add(1)
568
>
go func(data map[string]string) {
569
>
_, err1 := m.CreateNamespace(
570
>
&persistencespb.NamespaceInfo{
571
>
Id: id,
572
>
Name: name,
573
>
State: state,
574
>
Description: description,
575
>
Owner: owner,
576
>
Data: data,
577
>
},
578
>
&persistencespb.NamespaceConfig{
579
>
Retention: timestamp.DurationFromDays(retention),
580
>
HistoryArchivalState: historyArchivalState,
581
>
HistoryArchivalUri: historyArchivalURI,
582
>
VisibilityArchivalState: visibilityArchivalState,
583
>
VisibilityArchivalUri: visibilityArchivalURI,
584
>
BadBinaries: testBinaries,
585
>
},
586
>
&persistencespb.NamespaceReplicationConfig{
587
>
ActiveClusterName: clusterActive,
588
>
Clusters: clusters,
589
>
},
590
>
isGlobalNamespace,
591
>
configVersion,
592
>
failoverVersion,
593
>
)
594
>
if err1 == nil {
595
>
atomic.AddInt32(&successCount, 1)
596
>
}
597
>
wg.Done()
598
}(map[string]string{"k0": newValue})
599
}
601
>
m.Equal(int32(1), successCount)
602
>
603
>
resp, err3 := m.GetNamespace("", name)
604
>
m.NoError(err3)
605
>
m.NotNil(resp)
606
>
m.Equal(name, resp.Namespace.Info.Name)
607
>
m.Equal(state, resp.Namespace.Info.State)
608
>
m.Equal(description, resp.Namespace.Info.Description)
609
>
m.Equal(owner, resp.Namespace.Info.Owner)
610
>
m.EqualValues(time.Duration(retention)*time.Hour*24, resp.Namespace.Config.Retention.AsDuration())
611
>
m.Equal(historyArchivalState, resp.Namespace.Config.HistoryArchivalState)
612
>
m.Equal(historyArchivalURI, resp.Namespace.Config.HistoryArchivalUri)
613
>
m.Equal(visibilityArchivalState, resp.Namespace.Config.VisibilityArchivalState)
614
>
m.Equal(visibilityArchivalURI, resp.Namespace.Config.VisibilityArchivalUri)
615
>
m.ProtoEqual(testBinaries, resp.Namespace.Config.BadBinaries)
616
>
m.Equal(clusterActive, resp.Namespace.ReplicationConfig.ActiveClusterName)
617
>
m.Equal(len(clusters), len(resp.Namespace.ReplicationConfig.Clusters))
618
>
for index := range clusters {
619
>
m.Equal(clusters[index], resp.Namespace.ReplicationConfig.Clusters[index])
620
>
}
621
>
m.Equal(isGlobalNamespace, resp.IsGlobalNamespace)
622
>
m.Equal(configVersion, resp.Namespace.ConfigVersion)
623
>
m.Equal(failoverVersion, resp.Namespace.FailoverVersion)
624
>
625
>
// check namespace data
626
>
ss := strings.Split(resp.Namespace.Info.Data["k0"], "-")
627
>
m.Equal(2, len(ss))
628
>
vi, err := strconv.Atoi(ss[1])
629
>
m.NoError(err)
630
>
m.Equal(true, vi > 0 && vi <= concurrency)
631
}
632