func CreateTopics(t *testing.T, nbTopic int, nbPartitions []int, prefix string) []string {
	topics := make([]string, nbTopic)
	adm, err := confluent.NewAdminClient(&confluent.ConfigMap{
		"bootstrap.servers": helper.GetDefaultBootstrapServers(),
	})
	if err != nil {
		t.Fatalf("failed to create admin client: %v\n", err)
	}
	defer adm.Close()
	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()
	replicationFactor := 1
	specs := make([]confluent.TopicSpecification, nbTopic)
	for i := 0; i < len(specs); i++ {
		topics[i] = RandomTopicName(prefix)
		specs[i] = confluent.TopicSpecification{
			Topic:             topics[i],
			NumPartitions:     nbPartitions[i],
			ReplicationFactor: replicationFactor}
	}
	results, err := adm.CreateTopics(ctx, specs, confluent.SetAdminOperationTimeout(adminTimeout))
	if err != nil {
		t.Fatalf("failed to create topics %v: %v", topics, err)
	}
	for _, result := range results {
		t.Logf("%s\n", result)
	}
	return topics
}
Creates topics based on the array of nbPartitions param
package test
import (
	"bytes"
	"context"
	"crypto/rand"
	"fmt"
	"math/big"
	"reflect"
	"strconv"
	"testing"
	"time"
	confluent "github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/etf1/kafka-message-scheduler/internal/helper"
	"github.com/etf1/kafka-message-scheduler/schedule/kafka"
	kafka_store "github.com/etf1/kafka-message-scheduler/store/kafka"
)
const (
	adminTimeout   = 120 * time.Second
	timeout        = 10 * time.Second
	randomN        = 1000000
	pollTimeoutMs  = 100
	flushTimeoutMs = 10000
)
func NewKafkaStore(t *testing.T, nbTopic int, nbPartitions []int) (store *kafka_store.Store, topics []string) {
	topics = CreateTopics(t, nbTopic, nbPartitions, "scheduler")
	return NewKafkaStoreFromTopics(t, topics), topics
}
func NewKafkaStoreFromTopics(t *testing.T, topics []string) *kafka_store.Store {
	sessionTimeout := 6000
	store, err := kafka_store.NewStore(helper.GetDefaultBootstrapServers(), topics, "scheduler-cg", sessionTimeout)
	if err != nil {
		t.Fatalf("failed to create kafka store: %v\n", err)
	}
	return store
}