func TestMetadataRepoForIndexDefn(t *testing.T) {

	logging.SetLogLevel(logging.Trace)

	gometaL.LogEnable()
	gometaL.SetLogLevel(gometaL.LogLevelTrace)
	gometaL.SetPrefix("Indexing/Gometa")

	logging.Infof("Start TestMetadataRepo *********************************************************")

	/*
		var addr = "localhost:9885"
		var leader = "localhost:9884"

		repo, err := manager.NewMetadataRepo(addr, leader, "./config.json", nil)
		if err != nil {
			t.Fatal(err)
		}
		runTest(repo, t)
	*/

	os.MkdirAll("./data/", os.ModePerm)
	repo, _, err := manager.NewLocalMetadataRepo("localhost:5002", nil, nil, "./data/MetadataStore")
	if err != nil {
		t.Fatal(err)
	}
	runTest(repo, t)
}

// NewGSIIndexer manage new set of indexes under namespace->keyspace,
// also called as, pool->bucket.
// will return an error when,
// - GSI cluster is not available.
// - network partitions / errors.
func NewGSIIndexer(
	clusterURL, namespace, keyspace string) (datastore.Indexer, errors.Error) {

	l.SetLogLevel(l.Info)

	gsi := &gsiKeyspace{
		clusterURL:     clusterURL,
		namespace:      namespace,
		keyspace:       keyspace,
		indexes:        make(map[uint64]*secondaryIndex), // defnID -> index
		primaryIndexes: make(map[uint64]*secondaryIndex),
	}
	gsi.logPrefix = fmt.Sprintf("GSIC[%s; %s]", namespace, keyspace)

	// get the singleton-client
	client, err := getSingletonClient(clusterURL)
	if err != nil {
		l.Errorf("%v GSI instantiation failed: %v", gsi.logPrefix, err)
		return nil, errors.NewError(err, "GSI client instantiation failed")
	}
	gsi.gsiClient = client
	// refresh indexes for this service->namespace->keyspace
	if err := gsi.Refresh(); err != nil {
		l.Errorf("%v Refresh() failed: %v", gsi.logPrefix, err)
		return nil, err
	}
	l.Debugf("%v instantiated ...", gsi.logPrefix)
	return gsi, nil
}

func main() {
	logging.SetLogLevel(logging.Warn)
	runtime.GOMAXPROCS(runtime.NumCPU())

	cmdOptions, _, fset, err := querycmd.ParseArgs(os.Args[1:])
	if err != nil {
		logging.Fatalf("%v\n", err)
		os.Exit(1)
	} else if cmdOptions.Help || len(cmdOptions.OpType) < 1 {
		usage(fset)
		os.Exit(0)
	}

	config := c.SystemConfig.SectionConfig("queryport.client.", true)
	client, err := qclient.NewGsiClient(cmdOptions.Server, config)
	if err != nil {
		logging.Fatalf("%v\n", err)
		os.Exit(1)
	}

	if err = querycmd.HandleCommand(client, cmdOptions, false, os.Stdout); err != nil {
		fmt.Fprintf(os.Stderr, "Error occured %v\n", err)
	}
	client.Close()
}

func main() {
	logging.SetLogLevel(logging.Error)
	runtime.GOMAXPROCS(runtime.NumCPU())

	cmdOptions, args, fset, err := querycmd.ParseArgs(os.Args[1:])
	if err != nil {
		logging.Fatalf("%v", err)
		os.Exit(0)
	} else if cmdOptions.Help {
		usage(fset)
		os.Exit(0)
	} else if len(args) < 1 {
		logging.Fatalf("%v", "specify a command")
	}

	b, err := c.ConnectBucket(cmdOptions.Server, "default", "default")
	if err != nil {
		log.Fatal(err)
	}
	defer b.Close()
	maxvb, err := c.MaxVbuckets(b)
	if err != nil {
		log.Fatal(err)
	}

	config := c.SystemConfig.SectionConfig("queryport.client.", true)
	client, err := qclient.NewGsiClient(cmdOptions.Server, config)
	if err != nil {
		log.Fatal(err)
	}

	switch args[0] {
	case "sanity":
		err = doSanityTests(cmdOptions.Server, client)
		if err != nil {
			fmt.Fprintf(os.Stderr, "Error occured %v\n", err)
		}

	case "mb14786":
		err = doMB14786(cmdOptions.Server, client)
		if err != nil {
			fmt.Fprintf(os.Stderr, "Error occured %v\n", err)
		}

	case "mb13339":
		err = doMB13339(cmdOptions.Server, client)
		if err != nil {
			fmt.Fprintf(os.Stderr, "Error occured %v\n", err)
		}

	case "benchmark":
		doBenchmark(cmdOptions.Server, "localhost:8101")

	case "consistency":
		doConsistency(cmdOptions.Server, maxvb, client)
	}
	client.Close()
}

func argParse() string {
	buckets := "default"
	endpoints := "localhost:9020"

	flag.StringVar(&buckets, "buckets", buckets,
		"buckets to project")
	flag.StringVar(&endpoints, "endpoints", endpoints,
		"endpoints for mutations stream")
	flag.StringVar(&options.coordEndpoint, "coorendp", "localhost:9021",
		"coordinator endpoint")
	flag.IntVar(&options.stat, "stat", 1000,
		"periodic timeout to print dataport statistics")
	flag.IntVar(&options.timeout, "timeout", 0,
		"timeout for dataport to exit")
	flag.IntVar(&options.maxVbnos, "maxvb", 1024,
		"max number of vbuckets")
	flag.StringVar(&options.auth, "auth", "Administrator:asdasd",
		"Auth user and password")
	flag.BoolVar(&options.projector, "projector", false,
		"start projector for debug mode")
	flag.BoolVar(&options.debug, "debug", false,
		"run in debug mode")
	flag.BoolVar(&options.trace, "trace", false,
		"run in trace mode")

	flag.Parse()

	options.buckets = strings.Split(buckets, ",")
	options.endpoints = strings.Split(endpoints, ",")
	if options.debug {
		logging.SetLogLevel(logging.Debug)
	} else if options.trace {
		logging.SetLogLevel(logging.Trace)
	} else {
		logging.SetLogLevel(logging.Info)
	}

	args := flag.Args()
	if len(args) < 1 {
		usage()
		os.Exit(1)
	}
	return args[0]
}

func TestStreamBegin(t *testing.T) {
	maxBuckets, maxvbuckets, mutChanSize := 2, 8, 1000
	logging.SetLogLevel(logging.Silent)

	// start server
	appch := make(chan interface{}, mutChanSize)
	prefix := "indexer.dataport."
	config := c.SystemConfig.SectionConfig(prefix, true /*trim*/)
	daemon, err := NewServer(addr, maxvbuckets, config, appch)
	if err != nil {
		t.Fatal(err)
	}

	// start client
	flags := transport.TransportFlag(0).SetProtobuf()
	prefix = "projector.dataport.client."
	config = c.SystemConfig.SectionConfig(prefix, true /*trim*/)
	client, _ := NewClient(
		"cluster", "backfill", addr, flags, maxvbuckets, config)
	vbmaps := makeVbmaps(maxvbuckets, maxBuckets) // vbmaps
	for i := 0; i < maxBuckets; i++ {
		if err := client.SendVbmap(vbmaps[i]); err != nil {
			t.Fatal(err)
		}
	}

	// test a live StreamBegin
	bucket, vbno, vbuuid := "default0", uint16(maxvbuckets), uint64(1111)
	uuid := c.StreamID(bucket, vbno)
	vals, err := client.Getcontext()
	if err != nil {
		t.Fatal(err)
	}
	vbChans := vals[0].(map[string]chan interface{})
	if _, ok := vbChans[uuid]; ok {
		t.Fatal("duplicate id")
	}
	vb := c.NewVbKeyVersions(bucket, vbno, vbuuid, 1)
	seqno, docid, maxCount := uint64(10), []byte("document-name"), 10
	kv := c.NewKeyVersions(seqno, docid, maxCount)
	kv.AddStreamBegin()
	vb.AddKeyVersions(kv)
	err = client.SendKeyVersions([]*c.VbKeyVersions{vb}, true)
	client.Getcontext() // syncup
	if err != nil {
		t.Fatal(err)
	} else if _, ok := vbChans[uuid]; !ok {
		fmt.Printf("%v %v\n", len(vbChans), uuid)
		t.Fatal("failed StreamBegin")
	}
	client.Close()
	daemon.Close()
}

func TestStreamMgr_Timer(t *testing.T) {

	logging.SetLogLevel(logging.Trace)
	util.TT = t

	old_value := manager.NUM_VB
	manager.NUM_VB = 16
	defer func() { manager.NUM_VB = old_value }()

	// Running test
	runTimerTest()
}

// ResetConfig accepts a full-set or subset of global configuration
// and updates projector related fields.
func (p *Projector) ResetConfig(config c.Config) {
	p.rw.Lock()
	defer p.rw.Unlock()
	defer logging.Infof("%v\n", c.LogRuntime())

	// reset configuration.
	if cv, ok := config["projector.settings.log_level"]; ok {
		logging.SetLogLevel(logging.Level(cv.String()))
	}
	if cv, ok := config["projector.maxCpuPercent"]; ok {
		c.SetNumCPUs(cv.Int())
	}
	p.config = p.config.Override(config)

	// CPU-profiling
	cpuProfile, ok := config["projector.cpuProfile"]
	if ok && cpuProfile.Bool() && p.cpuProfFd == nil {
		cpuProfFname, ok := config["projector.cpuProfFname"]
		if ok {
			fname := cpuProfFname.String()
			logging.Infof("%v cpu profiling => %q\n", p.logPrefix, fname)
			p.cpuProfFd = p.startCPUProfile(fname)

		} else {
			logging.Errorf("Missing cpu-profile o/p filename\n")
		}

	} else if ok && !cpuProfile.Bool() {
		if p.cpuProfFd != nil {
			pprof.StopCPUProfile()
			logging.Infof("%v cpu profiling stopped\n", p.logPrefix)
		}
		p.cpuProfFd = nil

	} else if ok {
		logging.Warnf("%v cpu profiling already active !!\n", p.logPrefix)
	}

	// MEM-profiling
	memProfile, ok := config["projector.memProfile"]
	if ok && memProfile.Bool() {
		memProfFname, ok := config["projector.memProfFname"]
		if ok {
			fname := memProfFname.String()
			if p.takeMEMProfile(fname) {
				logging.Infof("%v mem profile => %q\n", p.logPrefix, fname)
			}
		} else {
			logging.Errorf("Missing mem-profile o/p filename\n")
		}
	}
}

func argParse() string {
	var buckets string

	flag.StringVar(&buckets, "buckets", "default",
		"buckets to listen")
	flag.IntVar(&options.maxVbno, "maxvb", 1024,
		"maximum number of vbuckets")
	flag.IntVar(&options.stats, "stats", 1000,
		"periodic timeout in mS, to print statistics, `0` will disable stats")
	flag.BoolVar(&options.printflogs, "flogs", false,
		"display failover logs")
	flag.StringVar(&options.auth, "auth", "",
		"Auth user and password")
	flag.BoolVar(&options.info, "info", false,
		"display informational logs")
	flag.BoolVar(&options.debug, "debug", false,
		"display debug logs")
	flag.BoolVar(&options.trace, "trace", false,
		"display trace logs")

	flag.Parse()

	options.buckets = strings.Split(buckets, ",")
	if options.debug {
		logging.SetLogLevel(logging.Debug)
	} else if options.trace {
		logging.SetLogLevel(logging.Trace)
	} else {
		logging.SetLogLevel(logging.Info)
	}

	args := flag.Args()
	if len(args) < 1 {
		usage()
		os.Exit(1)
	}
	return args[0]
}

func TestRequestHandler(t *testing.T) {

	logging.SetLogLevel(logging.Trace)

	cfg := common.SystemConfig.SectionConfig("indexer", true /*trim*/)
	cfg.Set("storage_dir", common.ConfigValue{"./data/", "metadata file path", "./"})
	os.MkdirAll("./data/", os.ModePerm)

	logging.Infof("Start TestRequestHandler *********************************************************")

	var config = "./config.json"

	logging.Infof("********** Setup index manager")
	var msgAddr = "localhost:9884"
	var httpAddr = "localhost:9102"
	addrPrv := util.NewFakeAddressProvider(msgAddr, httpAddr)
	mgr, err := manager.NewIndexManagerInternal(addrPrv, nil, cfg)
	if err != nil {
		t.Fatal(err)
	}
	mgr.StartCoordinator(config)
	defer mgr.Close()
	time.Sleep(time.Duration(1000) * time.Millisecond)

	logging.Infof("********** Start HTTP Server")
	go func() {
		if err := http.ListenAndServe(":9102", nil); err != nil {
			t.Fatal("Fail to start HTTP server on :9102")
		}
	}()

	logging.Infof("********** Cleanup Old Test")
	cleanupRequestHandlerTest(mgr, t)
	time.Sleep(time.Duration(1000) * time.Millisecond)

	logging.Infof("********** Start running request handler test")
	createIndexRequest(t)
	dropIndexRequest(t)

	logging.Infof("********** Cleanup Test")
	cleanupRequestHandlerTest(mgr, t)
	mgr.CleanupTopology()
	mgr.CleanupStabilityTimestamp()
	time.Sleep(time.Duration(1000) * time.Millisecond)

	logging.Infof("Done TestRequestHandler. Tearing down *********************************************************")
	mgr.Close()
	time.Sleep(time.Duration(1000) * time.Millisecond)
}

func argParse() []string {
	flag.StringVar(&options.auth, "auth", "",
		"Auth user and password")
	flag.BoolVar(&options.debug, "debug", false,
		"run in debug mode")
	flag.BoolVar(&options.trace, "trace", false,
		"run in trace mode")

	flag.Parse()

	if options.debug {
		logging.SetLogLevel(logging.Debug)
	} else if options.trace {
		logging.SetLogLevel(logging.Trace)
	} else {
		logging.SetLogLevel(logging.Info)
	}

	args := flag.Args()
	if len(args) < 1 {
		os.Exit(1)
	}
	return strings.Split(args[0], ",")
}

func TestStreamMgr_Monitor(t *testing.T) {

	logging.SetLogLevel(logging.Trace)
	util.TT = t

	old_value := manager.NUM_VB
	manager.NUM_VB = 16
	defer func() { manager.NUM_VB = old_value }()

	old_interval := manager.MONITOR_INTERVAL
	manager.MONITOR_INTERVAL = time.Duration(1000) * time.Millisecond
	defer func() { manager.MONITOR_INTERVAL = old_interval }()

	// Running test
	runMonitorTest()
}

func init() {
	log.Printf("In init()")
	logging.SetLogLevel(logging.Warn)

	var configpath string
	flag.StringVar(&configpath, "cbconfig", "../config/clusterrun_conf.json", "Path of the configuration file with data about Couchbase Cluster")
	flag.Parse()
	clusterconfig = tc.GetClusterConfFromFile(configpath)
	kvaddress = clusterconfig.KVAddress
	indexManagementAddress = clusterconfig.KVAddress
	indexScanAddress = clusterconfig.KVAddress
	seed = 1
	proddir, bagdir = tc.FetchMonsterToolPath()

	// setup cbauth
	if _, err := cbauth.InternalRetryDefaultInit(kvaddress, clusterconfig.Username, clusterconfig.Password); err != nil {
		log.Fatalf("Failed to initialize cbauth: %s", err)
	}
	secondaryindex.CheckCollation = true
	e := secondaryindex.DropAllSecondaryIndexes(indexManagementAddress)
	tc.HandleError(e, "Error in DropAllSecondaryIndexes")

	time.Sleep(5 * time.Second)
	// Working with Users10k and Users_mut dataset.
	u, _ := user.Current()
	dataFilePath = filepath.Join(u.HomeDir, "testdata/Users10k.txt.gz")
	mutationFilePath = filepath.Join(u.HomeDir, "testdata/Users_mut.txt.gz")
	tc.DownloadDataFile(tc.IndexTypesStaticJSONDataS3, dataFilePath, true)
	tc.DownloadDataFile(tc.IndexTypesMutationJSONDataS3, mutationFilePath, true)
	docs = datautility.LoadJSONFromCompressedFile(dataFilePath, "docid")
	mut_docs = datautility.LoadJSONFromCompressedFile(mutationFilePath, "docid")
	log.Printf("Emptying the default bucket")
	kvutility.EnableBucketFlush("default", "", clusterconfig.Username, clusterconfig.Password, kvaddress)
	kvutility.FlushBucket("default", "", clusterconfig.Username, clusterconfig.Password, kvaddress)
	time.Sleep(5 * time.Second)

	log.Printf("Create Index On the empty default Bucket()")
	var indexName = "index_eyeColor"
	var bucketName = "default"

	err := secondaryindex.CreateSecondaryIndex(indexName, bucketName, indexManagementAddress, "", []string{"eyeColor"}, false, nil, true, defaultIndexActiveTimeout, nil)
	tc.HandleError(err, "Error in creating the index")

	// Populate the bucket now
	log.Printf("Populating the default bucket")
	kvutility.SetKeyValues(docs, "default", "", clusterconfig.KVAddress)
}

func (gsi *gsiKeyspace) SetLogLevel(level qlog.Level) {
	switch level {
	case qlog.NONE:
		l.SetLogLevel(l.Silent)
	case qlog.SEVERE:
		l.SetLogLevel(l.Fatal)
	case qlog.ERROR:
		l.SetLogLevel(l.Error)
	case qlog.WARN:
		l.SetLogLevel(l.Warn)
	case qlog.INFO:
		l.SetLogLevel(l.Info)
	case qlog.REQUEST:
		l.SetLogLevel(l.Timing)
	case qlog.TRACE:
		l.SetLogLevel(l.Debug) //reversed
	case qlog.DEBUG:
		l.SetLogLevel(l.Trace)
	default:
		l.Warnf("Unknown query log level '%v'", level)
	}
}

func TestClient(t *testing.T) {
	maxBuckets, maxvbuckets, mutChanSize := 2, 8, 1000
	logging.SetLogLevel(logging.Silent)

	// start server
	appch := make(chan interface{}, mutChanSize)
	prefix := "indexer.dataport."
	config := c.SystemConfig.SectionConfig(prefix, true /*trim*/)
	daemon, err := NewServer(addr, maxvbuckets, config, appch)
	if err != nil {
		t.Fatal(err)
	}

	// start client and test number of connection.
	flags := transport.TransportFlag(0).SetProtobuf()
	prefix = "projector.dataport.client."
	config = c.SystemConfig.SectionConfig(prefix, true /*trim*/)
	maxconns := config["parConnections"].Int()
	client, err := NewClient(
		"cluster", "backfill", addr, flags, maxvbuckets, config)
	if err != nil {
		t.Fatal(err)
	} else if len(client.conns) != maxconns {
		t.Fatal("failed dataport client connections")
	} else if len(client.conns) != len(client.connChans) {
		t.Fatal("failed dataport client connection channels")
	} else if len(client.conns) != len(client.conn2Vbs) {
		t.Fatal("failed dataport client connection channels")
	} else {
		vbmaps := makeVbmaps(maxvbuckets, maxBuckets) // vbmaps
		for i := 0; i < maxBuckets; i++ {
			if err := client.SendVbmap(vbmaps[i]); err != nil {
				t.Fatal(err)
			}
		}
		validateClientInstance(client, maxvbuckets, maxconns, maxBuckets, t)
	}
	client.Close()
	daemon.Close()
}

func BenchmarkClientRequest(b *testing.B) {
	logging.SetLogLevel(logging.Silent)
	urlPrefix := common.SystemConfig["projector.adminport.urlPrefix"].String()
	client := NewHTTPClient(addr, urlPrefix)
	req := &testMessage{
		DefnID:          uint64(0x1234567812345678),
		Bucket:          "default",
		IsPrimary:       false,
		IName:           "example-index",
		Using:           "forrestdb",
		ExprType:        "n1ql",
		PartitionScheme: "simplekeypartition",
		Expression:      "x+1",
	}
	resp := &testMessage{}

	b.ResetTimer()
	for i := 0; i < b.N; i++ {
		if err := client.Request(req, resp); err != nil {
			b.Error(err)
		}
	}
}

func setLogger(config common.Config) {
	logLevel := config["indexer.settings.log_level"].String()
	level := logging.Level(logLevel)
	logging.Infof("Setting log level to %v", level)
	logging.SetLogLevel(level)
}

func init() {
	logging.SetLogLevel(logging.Silent)
	server = doServer("http://"+addr, q)
}

func TestCoordinator(t *testing.T) {

	logging.SetLogLevel(logging.Trace)

	logging.Infof("Start TestCoordinator *********************************************************")

	cfg := common.SystemConfig.SectionConfig("indexer", true /*trim*/)
	cfg.Set("storage_dir", common.ConfigValue{"./data/", "metadata file path", "./"})
	os.MkdirAll("./data/", os.ModePerm)

	/*
		var requestAddr = "localhost:9885"
		var leaderAddr = "localhost:9884"
	*/
	var config = "./config.json"
	manager.USE_MASTER_REPO = true
	defer func() { manager.USE_MASTER_REPO = false }()

	mgr, err := manager.NewIndexManagerInternal("localhost:9886", "localhost:"+manager.COORD_MAINT_STREAM_PORT, nil, cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer mgr.Close()
	mgr.StartCoordinator(config)
	time.Sleep(time.Duration(1000) * time.Millisecond)

	cleanup(mgr, t)
	time.Sleep(time.Duration(1000) * time.Millisecond)

	// Add a new index definition : 200
	idxDefn := &common.IndexDefn{
		DefnId:          common.IndexDefnId(200),
		Name:            "coordinator_test",
		Using:           common.ForestDB,
		Bucket:          "Default",
		IsPrimary:       false,
		SecExprs:        []string{"Testing"},
		ExprType:        common.N1QL,
		PartitionScheme: common.HASH,
		PartitionKey:    "Testing"}

	err = mgr.HandleCreateIndexDDL(idxDefn)
	if err != nil {
		t.Fatal(err)
	}
	time.Sleep(time.Duration(1000) * time.Millisecond)

	idxDefn, err = mgr.GetIndexDefnById(common.IndexDefnId(200))
	if err != nil {
		t.Fatal(err)
	}

	if idxDefn == nil {
		t.Fatal("Cannot find index definition")
	}

	topology, err := mgr.GetTopologyByBucket("Default")
	if err != nil {
		t.Fatal(err)
	}
	content, err := manager.MarshallIndexTopology(topology)
	if err != nil {
		t.Fatal(err)
	}
	logging.Infof("Topology after index creation : %s", string(content))

	inst := topology.GetIndexInstByDefn(common.IndexDefnId(200))
	if inst == nil || common.IndexState(inst.State) != common.INDEX_STATE_READY {
		t.Fatal("Index Inst not found for index defn 200 or inst state is not in READY")
	}

	cleanup(mgr, t)
	mgr.CleanupTopology()
	mgr.CleanupStabilityTimestamp()
	time.Sleep(time.Duration(1000) * time.Millisecond)

	logging.Infof("Done TestCoordinator. Tearing down *********************************************************")
	mgr.Close()
	time.Sleep(time.Duration(1000) * time.Millisecond)
}

func TestEventMgr(t *testing.T) {

	logging.SetLogLevel(logging.Trace)

	logging.Infof("Start TestEventMgr *********************************************************")

	cfg := common.SystemConfig.SectionConfig("indexer", true /*trim*/)
	cfg.Set("storage_dir", common.ConfigValue{"./data/", "metadata file path", "./"})
	os.MkdirAll("./data/", os.ModePerm)

	/*
		var requestAddr = "localhost:9885"
		var leaderAddr = "localhost:9884"
		var config = "./config.json"
	*/

	logging.Infof("Start Index Manager")
	factory := new(util.TestDefaultClientFactory)
	env := new(util.TestDefaultClientEnv)
	admin := manager.NewProjectorAdmin(factory, env, nil)
	//mgr, err := manager.NewIndexManagerInternal(requestAddr, leaderAddr, config, admin)
	mgr, err := manager.NewIndexManagerInternal("localhost:9886", "localhost:"+manager.COORD_MAINT_STREAM_PORT, admin, cfg)
	if err != nil {
		t.Fatal(err)
	}
	defer mgr.Close()

	cleanupEvtMgrTest(mgr, t)
	time.Sleep(time.Duration(1000) * time.Millisecond)

	logging.Infof("Start Listening to event")
	notifications, err := mgr.StartListenIndexCreate("TestEventMgr")
	if err != nil {
		t.Fatal(err)
	}

	// Add a new index definition : 300
	idxDefn := &common.IndexDefn{
		DefnId:          common.IndexDefnId(300),
		Name:            "event_mgr_test",
		Using:           common.ForestDB,
		Bucket:          "Default",
		IsPrimary:       false,
		SecExprs:        []string{"Testing"},
		ExprType:        common.N1QL,
		PartitionScheme: common.HASH,
		PartitionKey:    "Testing"}

	logging.Infof("Before DDL")
	err = mgr.HandleCreateIndexDDL(idxDefn)
	if err != nil {
		t.Fatal(err)
	}

	data := listen(notifications)
	if data == nil {
		t.Fatal("Does not receive notification from watcher")
	}

	idxDefn, err = common.UnmarshallIndexDefn(([]byte)(data))
	if err != nil {
		t.Fatal(err)
	}

	if idxDefn == nil {
		t.Fatal("Cannot unmarshall index definition")
	}

	if idxDefn.Name != "event_mgr_test" {
		t.Fatal("Index Definition Name mismatch")
	}

	cleanupEvtMgrTest(mgr, t)
	mgr.CleanupTopology()
	mgr.CleanupStabilityTimestamp()
	time.Sleep(time.Duration(1000) * time.Millisecond)

	logging.Infof("Stop TestEventMgr. Tearing down *********************************************************")

	mgr.Close()
	time.Sleep(time.Duration(1000) * time.Millisecond)
}