confluent kafka cluster create <name> --cloud <cloud provider> --region <cloud region>

confluent kafka cluster create dev0 --cloud aws --region us-east-1

confluent kafka topic create <name> --cluster <cluster ID>

confluent kafka topic create test_topic --cluster <ID of the dev0 cluster>

confluent kafka cluster list

confluent api-key create --resource <cluster ID>

|-- kafka-using-confluent/
	|-- consumer.go
	|-- producer.go
	|-- README.md
	|-- go.mod
	|-- go.sum

# Tạo folder để chứa source code
mkdir kafka-using-confluent
# Khởi tạo Go project
go mod init kafka-demo

touch producer.go

{
  "id": 5950,
  "quantity": 3,
  "item_type": "guitar",
  "price_per_unit": "BMg="
}

# producer.go
package main

import (
	"fmt"
	"os"
	"time"

	"github.com/confluentinc/confluent-kafka-go/kafka"
	"github.com/confluentinc/confluent-kafka-go/schemaregistry"
	"github.com/confluentinc/confluent-kafka-go/schemaregistry/serde"
	"github.com/confluentinc/confluent-kafka-go/schemaregistry/serde/jsonschema"
)

// Purchase is a simple record example
type Purchase struct {
	Id           int    `json:"id"`
	Quantity     int    `json:"quantity"`
	ItemType     string `json:"item_type"`
	PricePerUnit string `json:"price_per_unit"`
}

func main() {
	// Get env
	if len(os.Args) < 6 {
		fmt.Fprintf(os.Stderr, "Usage: %s <bootstrap-servers> <group> <topics..>\n", os.Args[0])
		os.Exit(1)
	}

	bootstrapServers := os.Args[1]
	clusterApiKey := os.Args[2]
	clusterApiSecret := os.Args[3]
	topic := os.Args[4]
	schemaRegistryUrl := os.Args[5]
	schemaRegistryApiKey := os.Args[6]
	schemaRegistryApiSecret := os.Args[7]

	hostname, err := os.Hostname()
	if err != nil {
		fmt.Printf("Failed to get hostname: %s", err)
		os.Exit(1)
	}

	// Create producer
	p, err := kafka.NewProducer(&kafka.ConfigMap{
		"bootstrap.servers": bootstrapServers,
		"sasl.mechanisms":   "PLAIN",
		"security.protocol": "SASL_SSL",
		"client.id":         hostname,
		"sasl.username":     clusterApiKey,
		"sasl.password":     clusterApiSecret,
		"acks":              "all",
	})
	if err != nil {
		fmt.Printf("Failed to create producer: %s", err)
		os.Exit(1)
	}

	// Create schema register client
	client, err := schemaregistry.NewClient(schemaregistry.NewConfigWithAuthentication(
		schemaRegistryUrl,
		schemaRegistryApiKey,
		schemaRegistryApiSecret,
	))
	if err != nil {
		fmt.Printf("Failed to create schema registry client: %s\n", err)
		os.Exit(1)
	}

	ser, err := jsonschema.NewSerializer(client, serde.ValueSerde, jsonschema.NewSerializerConfig())
	if err != nil {
		fmt.Printf("Failed to create serializer: %s\n", err)
		os.Exit(1)
	}

	// Serialize message
	msg := Purchase{
		Id:           5950,
		Quantity:     3,
		ItemType:     "guitar",
		PricePerUnit: "BMg=",
	}
	msgBytes, err := ser.Serialize(topic, &msg)
	if err != nil {
		fmt.Printf("Failed to serialize payload: %s\n", err)
		os.Exit(1)
	}

	// Asynchronous writes
	deliveryChan := make(chan kafka.Event, 10000)
	err = p.Produce(&kafka.Message{
		TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
		Value:          msgBytes,
		Key:            []byte(time.Now().UTC().Format(time.RFC1123)),
		Headers:        []kafka.Header{{Key: "myTestHeader", Value: []byte("header values are binary")}},
	}, deliveryChan)
	if err != nil {
		fmt.Printf("Failed to deliver message: %s", err)
		os.Exit(1)
	}

	// Go-routine to handle message delivery reports and possibly other event types (errors, stats, etc)
	go func() {
		for e := range p.Events() {
			switch ev := e.(type) {
			case *kafka.Message:
				if ev.TopicPartition.Error != nil {
					fmt.Printf("Failed to deliver: %v\n", ev.TopicPartition.Error)
				} else {
					fmt.Printf("Delivered message to topic %s partition [%d] @ offset %v\n",
						*ev.TopicPartition.Topic, ev.TopicPartition.Partition, ev.TopicPartition.Offset)
				}
			case kafka.Error:
				// Generic client instance-level errors, such as
				// broker connection failures, authentication issues, etc.
				//
				// These errors should generally be considered informational
				// as the underlying client will automatically try to
				// recover from any errors encountered, the application
				// does not need to take action on them.
				fmt.Printf("Error: %v\n", ev)
			default:
				fmt.Printf("Ignored event: %s\n", ev)
			}
		}
	}()

	// Wait for all messages to be delivered
	// Flush and close the producer and the events channel
	for p.Flush(1000) > 0 {
		fmt.Print("Still waiting to flush outstanding messages\n")
	}
	p.Close()
}

go run producer.go \
<bootstrap-servers> \
<cluster-api-key> \
<cluster-api-secret> \
<topic> \
<schema-registry-url> \
<schema-registry-api-key> \
<schema-registry-api-secret>

# consumer.go
package main

import (
	"fmt"
	"os"
	"os/signal"
	"syscall"

	"github.com/confluentinc/confluent-kafka-go/kafka"
)

func main() {
	if len(os.Args) < 5 {
		fmt.Fprintf(os.Stderr, "Usage: %s <bootstrap-servers> <group> <topics..>\n", os.Args[0])
		os.Exit(1)
	}

	bootstrapServers := os.Args[1]
	clusterApiKey := os.Args[2]
	clusterApiSecret := os.Args[3]
	group := os.Args[4]
	topics := os.Args[5:]

	sigchan := make(chan os.Signal, 1)
	signal.Notify(sigchan, syscall.SIGINT, syscall.SIGTERM)

	c, err := kafka.NewConsumer(&kafka.ConfigMap{
		"bootstrap.servers": bootstrapServers,
		// Avoid connecting to IPv6 brokers:
		// This is needed for the ErrAllBrokersDown show-case below
		// when using localhost brokers on OSX, since the OSX resolver
		// will return the IPv6 addresses first.
		// You typically don't need to specify this configuration property.
		"broker.address.family": "v4",
		"sasl.mechanisms":       "PLAIN",
		"security.protocol":     "SASL_SSL",
		"sasl.username":         clusterApiKey,
		"sasl.password":         clusterApiSecret,
		"group.id":              group,
		"session.timeout.ms":    6000,
		// Start reading from the first message of each assigned
		// partition if there are no previously committed offsets
		// for this group.
		"auto.offset.reset": "earliest",
		// Whether we store offsets automatically.
		"enable.auto.offset.store": false,
	})
	if err != nil {
		fmt.Fprintf(os.Stderr, "Failed to create consumer: %s\n", err)
		os.Exit(1)
	}

	fmt.Printf("Created Consumer %v\n", c)

	if err := c.SubscribeTopics(topics, nil); err != nil {
		fmt.Fprintf(os.Stderr, "Failed to subcribe topics %s %s\n", topics, err)
		os.Exit(1)
	}

	run := true

	for run {
		select {
		case sig := <-sigchan:
			fmt.Printf("Caught signal %v: terminating\n", sig)
			run = false
		default:
			// Retrieve records one-by-one
			ev := c.Poll(100)
			if ev == nil {
				continue
			}

			switch e := ev.(type) {
			case *kafka.Message:
				// Process the message received.
				fmt.Printf("%% Message on %s:\n%s\n",
					e.TopicPartition, string(e.Value))
				if e.Headers != nil {
					fmt.Printf("%% Headers: %v\n", e.Headers)
				}

				// We can store the offsets of the messages manually or let
				// the library do it automatically based on the setting
				// enable.auto.offset.store. Once an offset is stored, the
				// library takes care of periodically committing it to the broker
				// if enable.auto.commit isn't set to false (the default is true).
				// By storing the offsets manually after completely processing
				// each message, we can ensure atleast once processing.
				if _, err := c.StoreMessage(e); err != nil {
					fmt.Fprintf(os.Stderr, "%% Error storing offset after message %s:\n", e.TopicPartition)
				}
			case kafka.Error:
				// Errors should generally be considered
				// informational, the client will try to automatically recover.
				// But in this example we choose to terminate the application if all brokers are down.
				fmt.Fprintf(os.Stderr, "%% Error: %v: %v\n", e.Code(), e)
				if e.Code() == kafka.ErrAllBrokersDown {
					run = false
				}
			default:
				fmt.Printf("Ignored %v\n", e)
			}
		}
	}

	fmt.Printf("Closing consumer\n")
	c.Close()
}

# Chạy consumer, truyền vào các tham số cần thiết
go run consumer.go \
<bootstrap-servers> \
<cluster-api-key> \
<cluster-api-secret> \
<consumer-group-id> \
<topic-1> \
<topic-2> \
...
<topic-N>