func (mgr *MetadataManager) CreateConnection() {
	if mgr.zkConn != nil {
		// Return if the connection is good already
		if mgr.zkConn.State() == zk.StateConnected ||
			mgr.zkConn.State() == zk.StateHasSession ||
			mgr.zkConn.State() == zk.StateConnecting {
			return
		}
		// Close the connection because it probably expired
		mgr.zkConn.Close()
	}

	conn, _, err := zk.Connect(mgr.zookeepers, time.Second)
	if err != nil {
		conn.Close()
		log.Panic(err)
	}
	bns := baseNamespace{}
	ns := makeSubSpace(makeSubSpace(makeSubSpace(bns, "riak"), "frameworks"), mgr.frameworkID)
	lockPath := makeSubSpace(ns, "lock")
	zkLock := zk.NewLock(conn, lockPath.GetZKPath(), zk.WorldACL(zk.PermAll))

	mgr.zkConn = conn
	mgr.namespace = ns
	mgr.zkLock = *zkLock
}

func connect() *zk.Conn {
	zksStr := os.Getenv("ZOOKEEPER_SERVERS")
	zks := strings.Split(zksStr, ",")
	conn, _, err := zk.Connect(zks, time.Second)
	must(err)
	return conn
}

func NewZookeeperClient(machines []string) (*Client, error) {
	c, _, err := zk.Connect(machines, time.Second) //*10)
	if err != nil {
		panic(err)
	}
	return &Client{c}, nil
}

func NewStormClient(app *ApplicationContext, cluster string) (*StormClient, error) {
	// here we share the timeout w/ global zk
	zkconn, _, err := zk.Connect(app.Config.Storm[cluster].Zookeepers, time.Duration(app.Config.Zookeeper.Timeout)*time.Second)
	if err != nil {
		return nil, err
	}

	client := &StormClient{
		app:            app,
		cluster:        cluster,
		conn:           zkconn,
		stormGroupLock: sync.RWMutex{},
		stormGroupList: make(map[string]bool),
	}

	// Now get the first set of offsets and start a goroutine to continually check them
	client.refreshConsumerGroups()
	client.stormRefreshTicker = time.NewTicker(time.Duration(client.app.Config.Lagcheck.StormGroupRefresh) * time.Second)
	go func() {
		for _ = range client.stormRefreshTicker.C {
			client.refreshConsumerGroups()
		}
	}()

	return client, nil
}

func (z *ZookeeperClient) Connect() {
	var err error
	z.zkConn, _, err = zk.Connect(z.config.Servers, z.config.RecvTimeout)
	if err != nil {
		fmt.Println(err)
	}
}

func (zr *ZookeeperReporter) Connect() (zk.State, error) {
	if zr.ZKConnection != nil {
		state := zr.ZKConnection.State()
		switch state {
		case zk.StateUnknown, zk.StateConnectedReadOnly, zk.StateExpired, zk.StateAuthFailed, zk.StateConnecting:
			{
				//Disconnect, and let Reconnection happen
				log.Warn("Zookeeper Connection is in BAD State [", state, "] Reconnect")
				zr.ZKConnection.Close()
			}
		case zk.StateConnected, zk.StateHasSession:
			{
				log.Debug("Zookeeper Connection of [", zr.ServiceName, "][", zr.InstanceID, "] connected(", state, "), nothing to do.")
				return state, nil
			}
		case zk.StateDisconnected:
			{
				log.Info("Reporter Connection is Disconnected -> Reconnection")
			}
		}
	}
	conn, _, err := zk.Connect(zr.ZKHosts, 10*time.Second)
	if err != nil {
		zr.ZKConnection = nil
		log.Warn("Unable to Connect to ZooKeeper (", err, ")")
		return zk.StateDisconnected, err
	}
	zr.ZKConnection = conn
	var zkLogger ZKDebugLogger
	zr.ZKConnection.SetLogger(zkLogger)
	zr.ZKConnection = conn
	state := zr.ZKConnection.State()
	return state, nil
}

func TestConnectToZooKeeper(t *testing.T) {
	c, err := ConnectToZooKeeper(15, time.Millisecond*500, func(url string) bool {
		conn, _, err := zk.Connect([]string{url}, time.Second)
		if err != nil {
			return false
		}
		defer conn.Close()

		// Verify that we can perform operations, and that it's
		// a clean slate (/zookeeper should be the only path)
		children, _, err := conn.Children("/")
		if err != nil {
			return false
		}
		if len(children) != 1 {
			return false
		}
		if children[0] != "zookeeper" {
			return false
		}

		return true
	})
	assert.Nil(t, err)
	defer c.KillRemove()
}

func GetPreviousFrameworkID(
	zkServers []string,
	zkChroot string,
	frameworkName string,
) (fwid string, err error) {
	request := func() (string, error) {
		c, _, err := zk.Connect(zkServers, RPC_TIMEOUT)
		if err != nil {
			return "", err
		}
		defer c.Close()
		rawData, _, err := c.Get(zkChroot + "/" + frameworkName + "_framework_id")
		return string(rawData), err
	}

	backoff := 1
	for retries := 0; retries < RPC_RETRIES; retries++ {
		fwid, err = request()
		if err == nil {
			return fwid, err
		}
		time.Sleep(time.Duration(backoff) * time.Second)
		backoff = int(math.Min(float64(backoff<<1), 8))
	}
	return "", err
}

func GetPreviousReconciliationInfo(
	zkServers []string,
	zkChroot string,
	frameworkName string,
) (recon map[string]string, err error) {
	request := func() (map[string]string, error) {
		c, _, err := zk.Connect(zkServers, RPC_TIMEOUT)
		if err != nil {
			return map[string]string{}, err
		}
		defer c.Close()
		rawData, _, err := c.Get(zkChroot + "/" + frameworkName + "_reconciliation")
		if err == zk.ErrNoNode {
			return map[string]string{}, nil
		}
		if err != nil {
			return map[string]string{}, err
		}
		reconciliationInfo := map[string]string{}
		err = json.Unmarshal(rawData, &reconciliationInfo)
		return reconciliationInfo, err
	}

	backoff := 1
	for retries := 0; retries < RPC_RETRIES; retries++ {
		recon, err = request()
		if err == nil {
			return recon, err
		}
		time.Sleep(time.Duration(backoff) * time.Second)
		backoff = int(math.Min(float64(backoff<<1), 8))
	}
	return recon, err
}

// DialZkTimeout dial the server, and wait up to timeout until connection
func DialZkTimeout(zkAddr string, baseTimeout time.Duration, connectTimeout time.Duration) (*ZkConn, <-chan zookeeper.Event, error) {
	servers, err := resolveZkAddr(zkAddr)
	if err != nil {
		return nil, nil, err
	}

	sem.Acquire()
	defer sem.Release()
	zconn, session, err := zookeeper.Connect(servers, baseTimeout)
	if err != nil {
		return nil, nil, err
	}

	// Wait for connection, with a timeout, skipping transition states
	timer := time.NewTimer(connectTimeout)
	for {
		select {
		case <-timer.C:
			zconn.Close()
			return nil, nil, context.DeadlineExceeded
		case event := <-session:
			switch event.State {
			case zookeeper.StateConnected:
				// success
				return &ZkConn{conn: zconn}, session, nil

			case zookeeper.StateAuthFailed:
				// fast fail this one
				zconn.Close()
				return nil, nil, fmt.Errorf("zk connect failed: StateAuthFailed")
			}
		}
	}
}

func createNode() {
	const Node = "/greet"

	conn, _, _ := zk.Connect([]string{zkServer}, time.Second*5)

	create := func() error {
		var err error
		// try creating ephemeral node
		_, err = conn.Create(Node, []byte(httpPort), zk.FlagEphemeral, zk.WorldACL(zk.PermAll))
		return err
	}

	if create() != nil {
		// watch ephemeral node event.
		another, _, eventChan, _ := conn.GetW(Node)
		fmt.Println("Now listen", string(another))
	loop:
		for {
			event := <-eventChan
			if event.Type == zk.EventNodeDeleted || event.Type.String() == "Unknown" {
				// retry creating ephemeral node
				if create() != nil {
					break loop
				}
			}
		}
	}
}

// DialZk dials a ZK server and waits for connection event. Returns a ZkConn
// encapsulating the zookeeper.Conn, and the zookeeper session event
// channel to monitor the connection
//
// The value for baseTimeout is used as a session timeout as well, and
// will be used to negotiate a 'good' value with the server. From
// reading the zookeeper source code, it has to be between 6 and 60
// seconds (2x and 20x the tickTime by default, with default tick time
// being 3 seconds).  min session time, max session time and ticktime
// can all be overwritten on the zookeeper server side, so these
// numbers may vary.
//
// Then this baseTimeout is used to compute other related timeouts:
// - connect timeout is 1/3 of baseTimeout
// - recv timeout is 2/3 of baseTimeout minus a ping time
// - send timeout is 1/3 of baseTimeout
// - we try to send a ping a least every baseTimeout / 3
//
// Note the baseTimeout has *nothing* to do with the time between we
// call Dial and the maximum time before we receive the event on the
// session. The library will actually try to re-connect in the background
// (after each timeout), and may *never* send an event if the TCP connections
// always fail. Use DialZkTimeout to enforce a timeout for the initial connect.
func DialZk(zkAddr string, baseTimeout time.Duration) (*ZkConn, <-chan zookeeper.Event, error) {
	servers, err := resolveZkAddr(zkAddr)
	if err != nil {
		return nil, nil, err
	}

	sem.Acquire()
	defer sem.Release()
	zconn, session, err := zookeeper.Connect(servers, baseTimeout)
	if err != nil {
		return nil, nil, err
	}

	// Wait for connection, possibly forever, skipping transition states
	for {
		event := <-session
		switch event.State {
		case zookeeper.StateConnected:
			// success
			return &ZkConn{conn: zconn}, session, nil

		case zookeeper.StateAuthFailed:
			// fast fail this one
			zconn.Close()
			return nil, nil, fmt.Errorf("zk connect failed: StateAuthFailed")
		}
	}
}

func GetConnect() (conn *zk.Conn, err error) {
	conn, _, err = zk.Connect(hosts, timeOut*time.Second)
	if err != nil {
		fmt.Println(err)
	}
	return
}

// a health check for zookeeper
func zkHealthCheck() error {

	start := time.Now()
	lastError := time.Now()
	minUptime := time.Second * 2
	timeout := time.Second * 30
	zookeepers := []string{"127.0.0.1:2181"}

	for {
		if conn, _, err := zk.Connect(zookeepers, time.Second*10); err == nil {
			conn.Close()
		} else {
			conn.Close()
			glog.V(1).Infof("Could not connect to zookeeper: %s", err)
			lastError = time.Now()
		}
		// make sure that service has been good for at least minUptime
		if time.Since(lastError) > minUptime {
			break
		}
		if time.Since(start) > timeout {
			return fmt.Errorf("Zookeeper did not respond.")
		}
		time.Sleep(time.Millisecond * 1000)
	}
	glog.Info("zookeeper container started, browser at http://localhost:12181/exhibitor/v1/ui/index.html")
	return nil
}

// NewZk connect zookeeper
func NewZk(zks []string) (*zk.Conn, error) {
	c, _, err := zk.Connect(zks, time.Second*1)
	if err != nil {
		return nil, err
	}
	return c, nil
}

func PersistFrameworkID(
	fwid *mesos.FrameworkID,
	zkServers []string,
	zkChroot string,
	frameworkName string,
) error {
	c, _, err := zk.Connect(zkServers, RPC_TIMEOUT)
	if err != nil {
		return err
	}
	defer c.Close()
	// attempt to create the path
	_, err = c.Create(
		zkChroot,
		[]byte(""),
		0,
		zk.WorldACL(zk.PermAll),
	)
	if err != nil && err != zk.ErrNodeExists {
		return err
	}
	// attempt to write framework ID to <path> / <frameworkName>
	_, err = c.Create(zkChroot+"/"+frameworkName+"_framework_id",
		[]byte(fwid.GetValue()),
		0,
		zk.WorldACL(zk.PermAll))
	// TODO(tyler) when err is zk.ErrNodeExists, cross-check value
	if err != nil {
		return err
	}
	log.Info("Successfully persisted Framework ID to zookeeper.")

	return nil
}

func (this *ServiceBase) Connect() (err error) {
	if this.Conn != nil {
		this.Conn.Close()
	}
	this.Conn, _, err = zk.Connect(this.Hosts, this.ConnTimeout)
	return err
}

// CreateTestData: Implement the zkTestCallbacks interface
// Initialize the steps
func (test *zkLeaderResignationTest) InitFunc() ([]time.Duration, error) {
	test.info.Println("InitFunc()")

	// Create the connection to Zookeeper
	zkConn, _, err := zk.Connect([]string{test.testSetup.zkURL}, test.testSetup.heartBeat)
	test.zkConn = zkConn
	test.wg = &sync.WaitGroup{}

	if err != nil {
		test.error.Printf("leaderResignationTest: Error in zk.Connect (%s): %v",
			test.testSetup.zkURL, err)
		return nil, err
	}

	// Create the election node in ZooKeeper
	_, err = test.zkConn.Create(test.testSetup.electionNode,
		[]byte("data"), 0, zk.WorldACL(zk.PermAll))
	if err != nil {
		test.error.Printf("leaderResignationTest: Error creating the election node (%s): %v",
			test.testSetup.electionNode, err)
		return nil, err
	}

	return []time.Duration{
		time.Second * 1,
		time.Second * 1,
		time.Second * 1,
		time.Second * 1,
		time.Second * 1,
	}, nil
}

func (c *Client) initZk() {
	zkclient, _, err := zk.Connect(c.zkHosts, time.Second*30)
	if err != nil {
		panic(err)
	}

	c.zkClient = zkclient

	res, _, _, err := c.zkClient.GetW(c.zkRoot + c.zkRootRegionPath)

	if err != nil {
		panic(err)
	}

	c.rootServer = c.decodeMeta(res)
	c.getRegionConnection(c.getServerName(c.rootServer))

	res, _, _, err = c.zkClient.GetW(c.zkRoot + "/master")

	if err != nil {
		panic(err)
	}

	c.masterServer = c.decodeMeta(res)
}

func connect(zksStr string) (*zk.Conn, <-chan zk.Event) {
	//	zksStr := os.Getenv("ZOOKEEPER_SERVERS")
	zks := strings.Split(zksStr, ",")
	conn, connEvtChan, err := zk.Connect(zks, 5*time.Second)
	must(err)
	return conn, connEvtChan
}