confluent local services changes for Confluent Platform 8.0 (#3082)

Author: sgagniere

cmd/docs/main.go

@@ -29,14 +29,6 @@ func main() {
 		panic(err)
 	}
 
-	tempCH, err := createTemporaryConfluentHome()
-	if err != nil {
-		panic(err)
-	}
-	defer func() {
-		_ = os.RemoveAll(tempCH)
-	}()
-
 	// Generate documentation for both subsets of commands: cloud and on-prem
 	configs := []*config.Config{
 		{CurrentContext: "Cloud", Contexts: map[string]*config.Context{"Cloud": {PlatformName: "https://confluent.cloud"}}},
@@ -107,22 +99,3 @@ func removeLineFromFile(line, file string) error {
 
 	return os.WriteFile(file, out, 0644)
 }
-
-func createTemporaryConfluentHome() (string, error) {
-	dir := filepath.Join(os.TempDir(), "ch")
-	if err := os.Setenv("CONFLUENT_HOME", dir); err != nil {
-		return "", err
-	}
-
-	path := filepath.Join(dir, "share/java/confluent-control-center/control-center-0.0.0.jar")
-
-	if err := os.MkdirAll(filepath.Dir(path), 0777); err != nil {
-		return "", err
-	}
-
-	if err := os.WriteFile(path, []byte{}, 0644); err != nil {
-		return "", err
-	}
-
-	return dir, nil
-}

cmd/lint/main.go

@@ -169,6 +169,7 @@ var properNouns = []string{
 	"Key Encryption Key",
 	"KEK",
 	"Kotlin",
+	"KRaft Controller",
 	"Ktor",
 	"Kubernetes",
 	"Node.js",
@@ -258,6 +259,7 @@ var vocabWords = []string{
 	"keychain",
 	"kms",
 	"kotlin",
+	"kraft",
 	"ksql",
 	"ksqldb",
 	"ktor",

internal/command.go

@@ -119,7 +119,7 @@ func NewConfluentCommand(cfg *config.Config) *cobra.Command {
 	cmd.AddCommand(iam.New(cfg, prerunner))
 	cmd.AddCommand(kafka.New(cfg, prerunner))
 	cmd.AddCommand(ksql.New(cfg, prerunner))
-	cmd.AddCommand(local.New(prerunner))
+	cmd.AddCommand(local.New(cfg, prerunner))
 	cmd.AddCommand(login.New(cfg, prerunner, ccloudClientFactory, mdsClientManager, loginCredentialsManager, loginOrganizationManager, authTokenHandler))
 	cmd.AddCommand(logout.New(cfg, prerunner, authTokenHandler))
 	cmd.AddCommand(network.New(prerunner))

internal/local/command.go

@@ -4,6 +4,7 @@ import (
 	"github.com/spf13/cobra"
 
 	pcmd "github.com/confluentinc/cli/v4/pkg/cmd"
+	"github.com/confluentinc/cli/v4/pkg/config"
 	"github.com/confluentinc/cli/v4/pkg/local"
 )
 
@@ -21,7 +22,7 @@ func NewLocalCommand(cmd *cobra.Command, prerunner pcmd.PreRunner) *command {
 	}
 }
 
-func New(prerunner pcmd.PreRunner) *cobra.Command {
+func New(cfg *config.Config, prerunner pcmd.PreRunner) *cobra.Command {
 	cmd := &cobra.Command{
 		Use:   "local",
 		Short: "Manage a local Confluent Platform development environment.",
@@ -35,7 +36,7 @@ func New(prerunner pcmd.PreRunner) *cobra.Command {
 
 	c.AddCommand(NewCurrentCommand(prerunner))
 	c.AddCommand(NewDestroyCommand(prerunner))
-	c.AddCommand(NewServicesCommand(prerunner))
+	c.AddCommand(NewServicesCommand(cfg, prerunner))
 	c.AddCommand(NewVersionCommand(prerunner))
 
 	return cmd

internal/local/command_service.go

@@ -17,10 +17,13 @@ import (
 	"golang.org/x/text/cases"
 	"golang.org/x/text/language"
 
+	"github.com/confluentinc/properties"
+
 	"github.com/confluentinc/cli/v4/pkg/cmd"
 	"github.com/confluentinc/cli/v4/pkg/errors"
 	"github.com/confluentinc/cli/v4/pkg/output"
 	"github.com/confluentinc/cli/v4/pkg/spinner"
+	"github.com/confluentinc/cli/v4/pkg/utils"
 )
 
 func NewServiceCommand(service string, prerunner cmd.PreRunner) *cobra.Command {
@@ -132,6 +135,13 @@ func (c *command) runServiceStartCommand(cmd *cobra.Command, _ []string) error {
 	}
 
 	for _, dependency := range services[service].startDependencies {
+		compatible, err := c.isCompatibleService(dependency)
+		if err != nil {
+			return err
+		}
+		if !compatible {
+			continue
+		}
 		if err := c.startService(dependency, ""); err != nil {
 			return err
 		}
@@ -187,6 +197,13 @@ func (c *command) runServiceStopCommand(cmd *cobra.Command, _ []string) error {
 	}
 
 	for _, dependency := range services[service].stopDependencies {
+		compatible, err := c.isCompatibleService(dependency)
+		if err != nil {
+			return err
+		}
+		if !compatible {
+			continue
+		}
 		if err := c.stopService(dependency); err != nil {
 			return err
 		}
@@ -242,7 +259,12 @@ func NewServiceVersionCommand(service string, prerunner cmd.PreRunner) *cobra.Co
 func (c *command) runServiceVersionCommand(cmd *cobra.Command, _ []string) error {
 	service := cmd.Parent().Name()
 
-	ver, err := c.ch.GetVersion(service)
+	zookeeperMode, err := c.isZookeeperMode()
+	if err != nil {
+		return err
+	}
+
+	ver, err := c.ch.GetVersion(service, zookeeperMode)
 	if err != nil {
 		return err
 	}
@@ -252,7 +274,7 @@ func (c *command) runServiceVersionCommand(cmd *cobra.Command, _ []string) error
 }
 
 func (c *command) startService(service, configFile string) error {
-	if err := c.checkJavaVersion(service); err != nil {
+	if err := c.checkJavaVersion(); err != nil {
 		return err
 	}
 
@@ -286,7 +308,12 @@ func (c *command) startService(service, configFile string) error {
 }
 
 func (c *command) configService(service, configFile string) error {
-	port, err := c.ch.ReadServicePort(service)
+	zookeeperMode, err := c.isZookeeperMode()
+	if err != nil {
+		return err
+	}
+
+	port, err := c.ch.ReadServicePort(service, zookeeperMode)
 	if err != nil {
 		if err.Error() != "no port specified" {
 			return err
@@ -297,7 +324,7 @@ func (c *command) configService(service, configFile string) error {
 
 	var data []byte
 	if configFile == "" {
-		data, err = c.ch.ReadServiceConfig(service)
+		data, err = c.ch.ReadServiceConfig(service, zookeeperMode)
 	} else {
 		data, err = os.ReadFile(configFile)
 	}
@@ -370,6 +397,11 @@ func (c *command) startProcess(service string) error {
 		return err
 	}
 
+	err = c.setupMetaProperties(service)
+	if err != nil {
+		return err
+	}
+
 	start := exec.Command(scriptFile, configFile)
 
 	logFile, err := c.cc.GetLogFile(service)
@@ -675,7 +707,7 @@ func (c *command) checkOSVersion() error {
 	return nil
 }
 
-func (c *command) checkJavaVersion(service string) error {
+func (c *command) checkJavaVersion() error {
 	java := filepath.Join(os.Getenv("JAVA_HOME"), "/bin/java")
 	if os.Getenv("JAVA_HOME") == "" {
 		out, err := exec.Command("which", "java").Output()
@@ -696,7 +728,7 @@ func (c *command) checkJavaVersion(service string) error {
 	re := regexp.MustCompile(`.+ version "([\d._]+)"`)
 	javaVersion := string(re.FindSubmatch(data)[1])
 
-	isValid, err := isValidJavaVersion(service, javaVersion)
+	isValid, err := isValidJavaVersion(javaVersion)
 	if err != nil {
 		return err
 	}
@@ -709,7 +741,7 @@ func (c *command) checkJavaVersion(service string) error {
 	return nil
 }
 
-func isValidJavaVersion(service, javaVersion string) (bool, error) {
+func isValidJavaVersion(javaVersion string) (bool, error) {
 	// 1.8.0_152 -> 8.0_152 -> 8.0
 	javaVersion = strings.TrimPrefix(javaVersion, "1.")
 	javaVersion = strings.Split(javaVersion, "_")[0]
@@ -726,10 +758,6 @@ func isValidJavaVersion(service, javaVersion string) (bool, error) {
 		return false, nil
 	}
 
-	if service == "zookeeper" || service == "kafka" {
-		return true, nil
-	}
-
 	return true, nil
 }
 
@@ -748,6 +776,8 @@ func writeServiceName(service string) string {
 	switch service {
 	case "kafka-rest":
 		return "Kafka REST"
+	case "kraft-controller":
+		return "KRaft Controller"
 	case "ksql-server":
 		return "ksqlDB Server"
 	case "zookeeper":
@@ -757,3 +787,77 @@ func writeServiceName(service string) string {
 		return cases.Title(language.Und).String(service)
 	}
 }
+
+func (c *command) setupMetaProperties(service string) error {
+	// Only KRaft Controller and Kafka need to set up meta.properties
+	if service != "kraft-controller" && service != "kafka" {
+		return nil
+	}
+
+	// This step is only valid when running in KRaft mode
+	zookeeperMode, err := c.isZookeeperMode()
+	if err != nil {
+		return err
+	}
+	if zookeeperMode {
+		return nil
+	}
+
+	dataDir, err := c.cc.GetDataDir(service)
+	if err != nil {
+		return err
+	}
+	var metaFile string
+	switch service {
+	case "kraft-controller":
+		metaFile = filepath.Join(dataDir, "kraft-controller-logs", "meta.properties")
+	case "kafka":
+		metaFile = filepath.Join(dataDir, "kraft-broker-logs", "meta.properties")
+	}
+	if utils.FileExists(metaFile) { // formatting the properties file twice results in an error
+		return nil
+	}
+
+	kafkaStorage, err := c.ch.GetFile("bin", "kafka-storage")
+	if err != nil {
+		return err
+	}
+
+	var uuid string
+	var ok bool
+	if service == "kraft-controller" {
+		out, err := exec.Command(kafkaStorage, "random-uuid").Output()
+		if err != nil {
+			return err
+		}
+		uuid = strings.TrimSuffix(string(out), "\n")
+	} else if service == "kafka" {
+		// read the uuid from the controller meta.properties file since the broker needs to use the same id
+		// this file should exist since the controller is a dependency of the broker, and hence started first
+		controllerDataDir, err := c.cc.GetDataDir("kraft-controller")
+		if err != nil {
+			return err
+		}
+		controllerMetaFile := filepath.Join(controllerDataDir, "kraft-controller-logs", "meta.properties")
+		controllerMetaProperties, err := properties.LoadFile(controllerMetaFile, properties.UTF8)
+		if err != nil {
+			return err
+		}
+		uuid, ok = controllerMetaProperties.Get("cluster.id")
+		if !ok {
+			return errors.New("unable to retrieve cluster id from KRaft controller meta.properties file")
+		}
+	}
+
+	configFile, err := c.cc.GetConfigFile(service)
+	if err != nil {
+		return err
+	}
+
+	kafkaStorageArgs := []string{"format", "-t", uuid, "-c", configFile}
+	if service == "kraft-controller" {
+		kafkaStorageArgs = append(kafkaStorageArgs, "-s")
+	}
+
+	return exec.Command(kafkaStorage, kafkaStorageArgs...).Run()
+}

internal/local/command_service_test.go

@@ -62,15 +62,15 @@ func TestIsValidJavaVersion(t *testing.T) {
 	var isValid bool
 	var err error
 
-	isValid, err = isValidJavaVersion("", "1.8.0_152")
+	isValid, err = isValidJavaVersion("1.8.0_152")
 	req.NoError(err)
 	req.True(isValid)
 
-	isValid, err = isValidJavaVersion("", "9.0.4")
+	isValid, err = isValidJavaVersion("9.0.4")
 	req.NoError(err)
 	req.False(isValid)
 
-	isValid, err = isValidJavaVersion("zookeeper", "13")
+	isValid, err = isValidJavaVersion("13")
 	req.NoError(err)
 	req.True(isValid)
 }