One step closer to implement in-place trucate so that component resta…

…rts are unnecessary.

core/session_pool.go

@@ -169,6 +169,16 @@ func (m *SessionPool) Recover(unflushedCh chan *shared.BatchBuffer) {
 	}
 }
 
+// Lock - Prevent operations while a table maintenance event is in progress
+func (m *SessionPool) Lock() {
+	m.sessPoolLock.Lock()
+}
+
+// Unlock - Allow operations after a table maintenance event is complete
+func (m *SessionPool) Unlock() {
+	m.sessPoolLock.Unlock()
+}
+
 // Metrics - Return pool size and usage.
 func (m *SessionPool) Metrics() (poolSize, inUse, pooled, maxUsed int) {
 

core/table.go

@@ -52,9 +52,11 @@ func LoadTable(path string, kvStore *shared.KVStore, name string, consulClient *
 	defer tableCacheLock.Unlock()
 	if t, ok := tableCache[name]; ok {
 		t.kvStore = kvStore
+		u.Debugf("Found table %s in cache.", name)
 		return t, nil
 	}
 
+	u.Debugf("Loading table %s.", name)
 	sch, err := shared.LoadSchema(path, name, consulClient)
 	if err != nil {
 		return nil, err

quanta-admin/truncate.go

@@ -4,6 +4,7 @@ import (
 	"fmt"
 	"github.com/disney/quanta/shared"
 	"github.com/hashicorp/consul/api"
+	"time"
 )
 
 // TruncateCmd - Truncate command
@@ -38,6 +39,9 @@ func (c *TruncateCmd) Run(ctx *Context) error {
 		return fmt.Errorf("updateModTimeForTable  error %v", err)
 	}
 
+	// Give consumers time to flush and restart.
+	time.Sleep(time.Second * 5)
+
 	err = nukeData(consulClient, ctx.Port, c.Table, "truncate", c.DropEnums)
 	if err != nil {
 		return err

quanta-kinesis-consumer/quanta-kinesis-consumer.go

@@ -271,7 +271,6 @@ func main() {
 	for  {
 		ctx, main.cancelFunc = context.WithCancel(context.Background())
 		scanErr := main.consumer.Scan(ctx, main.scanAndProcess)
-		time.Sleep(time.Second * 5)
 		main.destroy()
 		if err := main.eg.Wait(); err != nil {
 			u.Errorf("session error: %v", err)
@@ -301,6 +300,7 @@ func (m *Main) destroy() {
 	for _, v := range m.shardChannels {
 		close(v)
 	}
+	time.Sleep(time.Second * 5)  // Allow time for completion
 }
 
 func (m *Main) scanAndProcess(v *consumer.Record) error {
@@ -422,7 +422,6 @@ func (m *Main) recoverInflight(recoverFunc func(unflushedCh chan *shared.BatchBu
 
 func (m *Main) schemaChangeListener(e shared.SchemaChangeEvent) {
 
-	time.Sleep(time.Second * 2)
 	if m.cancelFunc != nil {
 		m.cancelFunc()
 	}

quanta-proxy/quanta-proxy.go

@@ -234,19 +234,16 @@ func schemaChangeListener(e shared.SchemaChangeEvent) {
 
 	core.ClearTableCache()
 	src.GetSessionPool().Recover(nil)
+	src.GetSessionPool().Lock()
+	defer src.GetSessionPool().Unlock()
 	switch e.Event {
 	case shared.Drop:
 		schema.DefaultRegistry().SchemaDrop("quanta", e.Table, lex.TokenTable)
 		log.Printf("Dropped table %s", e.Table)
 	case shared.Modify:
 		log.Printf("Truncated table %s", e.Table)
-		schema.DefaultRegistry().SchemaDrop("quanta", e.Table, lex.TokenTable)
-		var err error
-		src, err = source.NewQuantaSource("", consulAddr, *quantaPort, sessionPoolSize)
-		if err != nil {
-			u.Error(err)
-		}
-		schema.RegisterSourceAsSchema("quanta", src)
+		time.Sleep(time.Second * 5)
+		schema.DefaultRegistry().SchemaRefresh("quanta")
 	case shared.Create:
 		schema.DefaultRegistry().SchemaDrop("quanta", "quanta", lex.TokenSource)
 		var err error