Function TestMaxSubscriptionsPerTopic

pulsaradmin/pkg/admin/topic_test.go:818–863

(t *testing.T)

}

func TestMaxSubscriptionsPerTopic(t *testing.T) {
	randomName := newTopicName()
	topic := "persistent://public/default/" + randomName
	cfg := &config.Config{}
	admin, err := New(cfg)
	assert.NoError(t, err)
	assert.NotNil(t, admin)
	topicName, err := utils.GetTopicName(topic)
	assert.NoError(t, err)
	err = admin.Topics().Create(*topicName, 4)
	assert.NoError(t, err)

	// Get default max subscriptions per topic
	maxSubscriptions, err := admin.Topics().GetMaxSubscriptionsPerTopic(*topicName)
	assert.NoError(t, err)
	assert.Equal(t, -1, maxSubscriptions)

	// Set new max subscriptions per topic
	err = admin.Topics().SetMaxSubscriptionsPerTopic(*topicName, 100)
	assert.NoError(t, err)

	// topic policy is an async operation,
	// so we need to wait for a while to get current value
	assert.Eventually(
		t,
		func() bool {
			maxSubscriptions, err = admin.Topics().GetMaxSubscriptionsPerTopic(*topicName)
			return err == nil && maxSubscriptions == 100
		},
		10*time.Second,
		100*time.Millisecond,
	)

	// Remove max subscriptions per topic policy
	err = admin.Topics().RemoveMaxSubscriptionsPerTopic(*topicName)
	assert.NoError(t, err)
	assert.Eventually(
		t,
		func() bool {
			maxSubscriptions, err = admin.Topics().GetMaxSubscriptionsPerTopic(*topicName)
			return err == nil && maxSubscriptions == -1
		},
		10*time.Second,
		100*time.Millisecond,
	)
}

func TestSchemaValidationEnforced(t *testing.T) {
	randomName := newTopicName()