Merge branch 'feature/dubbo-2.7.5' into metadata_report

common/constant/key.go

@@ -224,6 +224,9 @@ const (
 	KEY_SEPARATOR      = ":"
 	DEFAULT_PATH_TAG   = "metadata"
 	KEY_REVISON_PREFIX = "revision"
+
+	// metadata service
+	METADATA_SERVICE_NAME = "org.apache.dubbo.metadata.MetadataService"
 )
 
 // HealthCheck Router

config_center/apollo/impl.go

@@ -25,7 +25,8 @@ import (
 )
 
 import (
-	"github.com/pkg/errors"
+	gxset "github.com/dubbogo/gost/container/set"
+	perrors "github.com/pkg/errors"
 	"github.com/zouyx/agollo"
 )
 
@@ -119,7 +120,7 @@ func getNamespaceName(namespace string, configFileFormat agollo.ConfigFileFormat
 func (c *apolloConfiguration) GetInternalProperty(key string, opts ...cc.Option) (string, error) {
 	config := agollo.GetConfig(c.appConf.NamespaceName)
 	if config == nil {
-		return "", errors.New(fmt.Sprintf("nothing in namespace:%s ", key))
+		return "", perrors.New(fmt.Sprintf("nothing in namespace:%s ", key))
 	}
 	return config.GetStringValue(key, ""), nil
 }
@@ -128,6 +129,16 @@ func (c *apolloConfiguration) GetRule(key string, opts ...cc.Option) (string, er
 	return c.GetInternalProperty(key, opts...)
 }
 
+// PublishConfig will publish the config with the (key, group, value) pair
+func (c *apolloConfiguration) PublishConfig(string, string, string) error {
+	return perrors.New("unsupport operation")
+}
+
+// GetConfigKeysByGroup will return all keys with the group
+func (c *apolloConfiguration) GetConfigKeysByGroup(group string) (*gxset.HashSet, error) {
+	return nil, perrors.New("unsupport operation")
+}
+
 func (c *apolloConfiguration) GetProperties(key string, opts ...cc.Option) (string, error) {
 	/**
 	 * when group is not null, we are getting startup configs(config file) from Config Center, for example:
@@ -135,7 +146,7 @@ func (c *apolloConfiguration) GetProperties(key string, opts ...cc.Option) (stri
 	 */
 	config := agollo.GetConfig(key)
 	if config == nil {
-		return "", errors.New(fmt.Sprintf("nothing in namespace:%s ", key))
+		return "", perrors.New(fmt.Sprintf("nothing in namespace:%s ", key))
 	}
 	return config.GetContent(agollo.Properties), nil
 }

config_center/dynamic_configuration.go

@@ -21,14 +21,18 @@ import (
 	"time"
 )
 
+import (
+	gxset "github.com/dubbogo/gost/container/set"
+)
+
 import (
 	"github.com/apache/dubbo-go/common"
 	"github.com/apache/dubbo-go/config_center/parser"
 )
 
-//////////////////////////////////////////
+// ////////////////////////////////////////
 // DynamicConfiguration
-//////////////////////////////////////////
+// ////////////////////////////////////////
 const (
 	// DEFAULT_GROUP: default group
 	DEFAULT_GROUP = "dubbo"
@@ -42,14 +46,20 @@ type DynamicConfiguration interface {
 	SetParser(parser.ConfigurationParser)
 	AddListener(string, ConfigurationListener, ...Option)
 	RemoveListener(string, ConfigurationListener, ...Option)
-	//GetProperties get properties file
+	// GetProperties get properties file
 	GetProperties(string, ...Option) (string, error)
 
-	//GetRule get Router rule properties file
+	// GetRule get Router rule properties file
 	GetRule(string, ...Option) (string, error)
 
-	//GetInternalProperty get value by key in Default properties file(dubbo.properties)
+	// GetInternalProperty get value by key in Default properties file(dubbo.properties)
 	GetInternalProperty(string, ...Option) (string, error)
+
+	// PublishConfig will publish the config with the (key, group, value) pair
+	PublishConfig(string, string, string) error
+
+	// GetConfigKeysByGroup will return all keys with the group
+	GetConfigKeysByGroup(group string) (*gxset.HashSet, error)
 }
 
 // Options ...
@@ -75,7 +85,7 @@ func WithTimeout(time time.Duration) Option {
 	}
 }
 
-//GetRuleKey The format is '{interfaceName}:[version]:[group]'
+// GetRuleKey The format is '{interfaceName}:[version]:[group]'
 func GetRuleKey(url common.URL) string {
 	return url.ColonSeparatedKey()
 }

config_center/mock_dynamic_config.go

@@ -22,6 +22,7 @@ import (
 )
 
 import (
+	gxset "github.com/dubbogo/gost/container/set"
 	"gopkg.in/yaml.v2"
 )
 
@@ -81,6 +82,16 @@ func (f *MockDynamicConfigurationFactory) GetDynamicConfiguration(_ *common.URL)
 
 }
 
+// PublishConfig will publish the config with the (key, group, value) pair
+func (c *MockDynamicConfiguration) PublishConfig(string, string, string) error {
+	return nil
+}
+
+// GetConfigKeysByGroup will return all keys with the group
+func (c *MockDynamicConfiguration) GetConfigKeysByGroup(group string) (*gxset.HashSet, error) {
+	return gxset.NewSet(c.content), nil
+}
+
 // MockDynamicConfiguration ...
 type MockDynamicConfiguration struct {
 	parser   parser.ConfigurationParser

config_center/nacos/impl.go

@@ -18,10 +18,12 @@
 package nacos
 
 import (
+	"strings"
 	"sync"
 )
 
 import (
+	gxset "github.com/dubbogo/gost/container/set"
 	"github.com/nacos-group/nacos-sdk-go/vo"
 	perrors "github.com/pkg/errors"
 )
@@ -74,7 +76,7 @@ func (n *nacosDynamicConfiguration) RemoveListener(key string, listener config_c
 	n.removeListener(key, listener)
 }
 
-//nacos distinguishes configuration files based on group and dataId. defalut group = "dubbo" and dataId = key
+// GetProperties nacos distinguishes configuration files based on group and dataId. defalut group = "dubbo" and dataId = key
 func (n *nacosDynamicConfiguration) GetProperties(key string, opts ...config_center.Option) (string, error) {
 	return n.GetRule(key, opts...)
 }
@@ -84,6 +86,33 @@ func (n *nacosDynamicConfiguration) GetInternalProperty(key string, opts ...conf
 	return n.GetProperties(key, opts...)
 }
 
+// PublishConfig will publish the config with the (key, group, value) pair
+func (n *nacosDynamicConfiguration) PublishConfig(key string, group string, value string) error {
+
+	group = n.resolvedGroup(group)
+
+	ok, err := (*n.client.Client()).PublishConfig(vo.ConfigParam{
+		DataId:  key,
+		Group:   group,
+		Content: value,
+	})
+
+	if err != nil {
+		return perrors.WithStack(err)
+	}
+	if !ok {
+		return perrors.New("publish config to Nocos failed")
+	}
+	return nil
+}
+
+// GetConfigKeysByGroup will return all keys with the group
+func (n *nacosDynamicConfiguration) GetConfigKeysByGroup(group string) (*gxset.HashSet, error) {
+	// TODO (the golang client of nacos does not support batch API)
+	// we should build a issue and then think about how to resolve this problem
+	return nil, perrors.New("unsupport operation, wait for implement")
+}
+
 // GetRule Get router rule
 func (n *nacosDynamicConfiguration) GetRule(key string, opts ...config_center.Option) (string, error) {
 	tmpOpts := &config_center.Options{}
@@ -92,12 +121,12 @@ func (n *nacosDynamicConfiguration) GetRule(key string, opts ...config_center.Op
 	}
 	content, err := (*n.client.Client()).GetConfig(vo.ConfigParam{
 		DataId: key,
-		Group:  tmpOpts.Group,
+		Group:  n.resolvedGroup(tmpOpts.Group),
 	})
 	if err != nil {
 		return "", perrors.WithStack(err)
 	} else {
-		return string(content), nil
+		return content, nil
 	}
 }
 
@@ -145,6 +174,15 @@ func (n *nacosDynamicConfiguration) Destroy() {
 	n.closeConfigs()
 }
 
+// resolvedGroup will regular the group. Now, it will replace the '/' with '-'.
+// '/' is a special character for nacos
+func (n *nacosDynamicConfiguration) resolvedGroup(group string) string {
+	if len(group) <= 0 {
+		return group
+	}
+	return strings.ReplaceAll(group, "/", "-")
+}
+
 // IsAvailable Get available status
 func (n *nacosDynamicConfiguration) IsAvailable() bool {
 	select {
@@ -155,12 +193,12 @@ func (n *nacosDynamicConfiguration) IsAvailable() bool {
 	}
 }
 
-func (r *nacosDynamicConfiguration) closeConfigs() {
-	r.cltLock.Lock()
-	client := r.client
-	r.client = nil
-	r.cltLock.Unlock()
+func (n *nacosDynamicConfiguration) closeConfigs() {
+	n.cltLock.Lock()
+	client := n.client
+	n.client = nil
+	n.cltLock.Unlock()
 	// Close the old client first to close the tmp node
 	client.Close()
-	logger.Infof("begin to close provider nacos client")
+	logger.Infof("begin to close provider n client")
 }

config_center/nacos/impl_test.go

@@ -60,12 +60,7 @@ func runMockConfigServer(configHandler func(http.ResponseWriter, *http.Request),
 
 func mockCommonNacosServer() *httptest.Server {
 	return runMockConfigServer(func(writer http.ResponseWriter, request *http.Request) {
-		data := `
-	dubbo.service.com.ikurento.user.UserProvider.cluster=failback
-	dubbo.service.com.ikurento.user.UserProvider.protocol=myDubbo1
-	dubbo.protocols.myDubbo.port=20000
-	dubbo.protocols.myDubbo.name=dubbo
-`
+		data := "true"
 		fmt.Fprintf(writer, "%s", data)
 	}, func(writer http.ResponseWriter, request *http.Request) {
 		data := `dubbo.properties%02dubbo%02dubbo.service.com.ikurento.user.UserProvider.cluster=failback`
@@ -93,6 +88,16 @@ func Test_GetConfig(t *testing.T) {
 	assert.NoError(t, err)
 }
 
+func TestNacosDynamicConfiguration_PublishConfig(t *testing.T) {
+	nacos, err := initNacosData(t)
+	assert.Nil(t, err)
+	key := "myKey"
+	group := "/custom/a/b"
+	value := "MyValue"
+	err = nacos.PublishConfig(key, group, value)
+	assert.Nil(t, err)
+}
+
 func Test_AddListener(t *testing.T) {
 	nacos, err := initNacosData(t)
 	assert.NoError(t, err)

config_center/nacos/listener.go

@@ -35,11 +35,11 @@ func callback(listener config_center.ConfigurationListener, namespace, group, da
 	listener.Process(&config_center.ConfigChangeEvent{Key: dataId, Value: data, ConfigType: remoting.EventTypeUpdate})
 }
 
-func (l *nacosDynamicConfiguration) addListener(key string, listener config_center.ConfigurationListener) {
-	_, loaded := l.keyListeners.Load(key)
+func (n *nacosDynamicConfiguration) addListener(key string, listener config_center.ConfigurationListener) {
+	_, loaded := n.keyListeners.Load(key)
 	if !loaded {
 		_, cancel := context.WithCancel(context.Background())
-		err := (*l.client.Client()).ListenConfig(vo.ConfigParam{
+		err := (*n.client.Client()).ListenConfig(vo.ConfigParam{
 			DataId: key,
 			Group:  "dubbo",
 			OnChange: func(namespace, group, dataId, data string) {
@@ -49,14 +49,14 @@ func (l *nacosDynamicConfiguration) addListener(key string, listener config_cent
 		logger.Errorf("nacos : listen config fail, error:%v ", err)
 		newListener := make(map[config_center.ConfigurationListener]context.CancelFunc)
 		newListener[listener] = cancel
-		l.keyListeners.Store(key, newListener)
+		n.keyListeners.Store(key, newListener)
 	} else {
 		// TODO check goroutine alive, but this version of go_nacos_sdk is not support.
 		logger.Infof("profile:%s. this profile is already listening", key)
 	}
 }
 
-func (l *nacosDynamicConfiguration) removeListener(key string, listener config_center.ConfigurationListener) {
+func (n *nacosDynamicConfiguration) removeListener(key string, listener config_center.ConfigurationListener) {
 	// TODO: not supported in current go_nacos_sdk version
 	logger.Warn("not supported in current go_nacos_sdk version")
 }

config_center/zookeeper/impl.go

@@ -25,6 +25,7 @@ import (
 
 import (
 	"github.com/dubbogo/go-zookeeper/zk"
+	gxset "github.com/dubbogo/gost/container/set"
 	perrors "github.com/pkg/errors"
 )
 
@@ -39,8 +40,9 @@ import (
 
 const (
 	// ZkClient
-	//zookeeper client name
-	ZkClient = "zk config_center"
+	// zookeeper client name
+	ZkClient      = "zk config_center"
+	pathSeparator = "/"
 )
 
 type zookeeperDynamicConfiguration struct {
@@ -143,11 +145,39 @@ func (c *zookeeperDynamicConfiguration) GetProperties(key string, opts ...config
 	return string(content), nil
 }
 
-//For zookeeper, getConfig and getConfigs have the same meaning.
+// GetInternalProperty For zookeeper, getConfig and getConfigs have the same meaning.
 func (c *zookeeperDynamicConfiguration) GetInternalProperty(key string, opts ...config_center.Option) (string, error) {
 	return c.GetProperties(key, opts...)
 }
 
+// PublishConfig will put the value into Zk with specific path
+func (c *zookeeperDynamicConfiguration) PublishConfig(key string, group string, value string) error {
+	path := c.getPath(key, group)
+	err := c.client.CreateWithValue(path, []byte(value))
+	if err != nil {
+		return perrors.WithStack(err)
+	}
+	return nil
+}
+
+// GetConfigKeysByGroup will return all keys with the group
+func (c *zookeeperDynamicConfiguration) GetConfigKeysByGroup(group string) (*gxset.HashSet, error) {
+	path := c.getPath("", group)
+	result, err := c.client.GetChildren(path)
+	if err != nil {
+		return nil, perrors.WithStack(err)
+	}
+
+	if len(result) == 0 {
+		return nil, perrors.New("could not find keys with group: " + group)
+	}
+	set := gxset.NewSet()
+	for _, e := range result {
+		set.Add(e)
+	}
+	return set, nil
+}
+
 func (c *zookeeperDynamicConfiguration) GetRule(key string, opts ...config_center.Option) (string, error) {
 	return c.GetProperties(key, opts...)
 }
@@ -214,3 +244,17 @@ func (c *zookeeperDynamicConfiguration) closeConfigs() {
 func (c *zookeeperDynamicConfiguration) RestartCallBack() bool {
 	return true
 }
+
+func (c *zookeeperDynamicConfiguration) getPath(key string, group string) string {
+	if len(key) == 0 {
+		return c.buildPath(group)
+	}
+	return c.buildPath(group) + pathSeparator + key
+}
+
+func (c *zookeeperDynamicConfiguration) buildPath(group string) string {
+	if len(group) == 0 {
+		group = config_center.DEFAULT_GROUP
+	}
+	return c.rootPath + pathSeparator + group
+}

config_center/zookeeper/impl_test.go

@@ -24,6 +24,7 @@ import (
 
 import (
 	"github.com/dubbogo/go-zookeeper/zk"
+	gxset "github.com/dubbogo/gost/container/set"
 	"github.com/stretchr/testify/assert"
 )
 
@@ -156,6 +157,26 @@ func Test_RemoveListener(t *testing.T) {
 	assert.Equal(t, "", listener.event)
 }
 
+func TestZookeeperDynamicConfiguration_PublishConfig(t *testing.T) {
+	value := "Test Data"
+	customGroup := "Custom Group"
+	key := "myKey"
+	ts, zk := initZkData(config_center.DEFAULT_GROUP, t)
+	defer ts.Stop()
+	err := zk.PublishConfig(key, customGroup, value)
+	assert.Nil(t, err)
+	result, err := zk.GetInternalProperty("myKey", config_center.WithGroup(customGroup))
+	assert.Nil(t, err)
+	assert.Equal(t, value, result)
+
+	var keys *gxset.HashSet
+	keys, err = zk.GetConfigKeysByGroup(customGroup)
+	assert.Nil(t, err)
+	assert.Equal(t, 1, keys.Size())
+	assert.True(t, keys.Contains(key))
+
+}
+
 type mockDataListener struct {
 	wg    sync.WaitGroup
 	event string

metadata/namemapping/dynamic/service_name_mapping.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dynamic
+
+import (
+	"strconv"
+	"time"
+)
+
+import (
+	"github.com/dubbogo/gost/container/set"
+	perrors "github.com/pkg/errors"
+)
+
+import (
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/config"
+	"github.com/apache/dubbo-go/config_center"
+	"github.com/apache/dubbo-go/metadata"
+)
+
+const (
+	defaultGroup = config_center.DEFAULT_GROUP
+	slash        = "/"
+)
+
+// DynamicConfigurationServiceNameMapping is the implementation based on config center
+type DynamicConfigurationServiceNameMapping struct {
+	dc config_center.DynamicConfiguration
+}
+
+// Map will map the service to this application-level service
+func (d *DynamicConfigurationServiceNameMapping) Map(serviceInterface string, group string, version string, protocol string) error {
+	// metadata service is admin service, should not be mapped
+	if constant.METADATA_SERVICE_NAME == serviceInterface {
+		return perrors.New("try to map the metadata service, will be ignored")
+	}
+
+	appName := config.GetApplicationConfig().Name
+	value := time.Now().UnixNano()
+
+	err := d.dc.PublishConfig(appName,
+		d.buildGroup(serviceInterface),
+		strconv.FormatInt(value, 10))
+	if err != nil {
+		return perrors.WithStack(err)
+	}
+	return nil
+}
+
+// Get will return the application-level services. If not found, the empty set will be returned.
+// if the dynamic configuration got error, the error will return
+func (d *DynamicConfigurationServiceNameMapping) Get(serviceInterface string, group string, version string, protocol string) (*gxset.HashSet, error) {
+	return d.dc.GetConfigKeysByGroup(d.buildGroup(serviceInterface))
+}
+
+// buildGroup will return group, now it looks like defaultGroup/serviceInterface
+func (d *DynamicConfigurationServiceNameMapping) buildGroup(serviceInterface string) string {
+	// the issue : https://github.com/apache/dubbo/issues/4671
+	// so other params are ignored and remove, including group string, version string, protocol string
+	return defaultGroup + slash + serviceInterface
+}
+
+// NewServiceNameMapping will create an instance of DynamicConfigurationServiceNameMapping
+func NewServiceNameMapping(dc config_center.DynamicConfiguration) metadata.ServiceNameMapping {
+	return &DynamicConfigurationServiceNameMapping{dc: dc}
+}

metadata/namemapping/dynamic/service_name_mapping_test.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dynamic
+
+import (
+	"testing"
+)
+
+import (
+	gxset "github.com/dubbogo/gost/container/set"
+	"github.com/stretchr/testify/assert"
+)
+
+import (
+	"github.com/apache/dubbo-go/common/constant"
+	"github.com/apache/dubbo-go/config"
+	"github.com/apache/dubbo-go/config_center"
+)
+
+func TestDynamicConfigurationServiceNameMapping(t *testing.T) {
+
+	// mock data
+	appName := "myApp"
+	dc, err := (&config_center.MockDynamicConfigurationFactory{
+		Content: appName,
+	}).GetDynamicConfiguration(nil)
+	config.GetApplicationConfig().Name = appName
+
+	mapping := NewServiceNameMapping(dc)
+	intf := constant.METADATA_SERVICE_NAME
+	group := "myGroup"
+	version := "myVersion"
+	protocol := "myProtocol"
+
+	err = mapping.Map(intf, group, version, protocol)
+	assert.NotNil(t, err)
+	intf = "MyService"
+	err = mapping.Map(intf, group, version, protocol)
+	assert.Nil(t, err)
+
+	var result *gxset.HashSet
+	result, err = mapping.Get(intf, group, version, protocol)
+	assert.Nil(t, err)
+	assert.Equal(t, 1, result.Size())
+	assert.True(t, result.Contains(appName))
+}

metadata/namemapping/memory/service_name_mapping.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package memory
+
+import (
+	gxset "github.com/dubbogo/gost/container/set"
+)
+
+import (
+	"github.com/apache/dubbo-go/config"
+)
+
+type InMemoryServiceNameMapping struct{}
+
+func (i InMemoryServiceNameMapping) Map(serviceInterface string, group string, version string, protocol string) error {
+	return nil
+}
+
+func (i InMemoryServiceNameMapping) Get(serviceInterface string, group string, version string, protocol string) (*gxset.HashSet, error) {
+	return gxset.NewSet(config.GetApplicationConfig().Name), nil
+}

metadata/service_name_mapping.go

+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements.  See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License.  You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package metadata
+
+import (
+	gxset "github.com/dubbogo/gost/container/set"
+)
+
+// ServiceNameMapping try to build the mapping between application-level service and interface-level service.
+type ServiceNameMapping interface {
+
+	// Map will map the service to this application-level service
+	Map(serviceInterface string, group string, version string, protocol string) error
+
+	// Get will return the application-level services
+	Get(serviceInterface string, group string, version string, protocol string) (*gxset.HashSet, error)
+}

remoting/zookeeper/client.go

@@ -392,8 +392,16 @@ func (z *ZookeeperClient) Close() {
 	logger.Warnf("zkClient{name:%s, zk addr:%s} exit now.", z.name, z.ZkAddrs)
 }
 
-// Create ...
+// Create will create the node recursively, which means that if the parent node is absent,
+// it will create parent node first.
+// And the value for the basePath is ""
 func (z *ZookeeperClient) Create(basePath string) error {
+	return z.CreateWithValue(basePath, []byte(""))
+}
+
+// CreateWithValue will create the node recursively, which means that if the parent node is absent,
+// it will create parent node first.
+func (z *ZookeeperClient) CreateWithValue(basePath string, value []byte) error {
 	var (
 		err     error
 		tmpPath string
@@ -407,7 +415,7 @@ func (z *ZookeeperClient) Create(basePath string) error {
 		conn := z.Conn
 		z.Unlock()
 		if conn != nil {
-			_, err = conn.Create(tmpPath, []byte(""), 0, zk.WorldACL(zk.PermAll))
+			_, err = conn.Create(tmpPath, value, 0, zk.WorldACL(zk.PermAll))
 		}
 
 		if err != nil {