Merge branch 'week_5'

.gitignore

@@ -6,6 +6,8 @@
 /week_3/bin/
 /week_5/bin/
 /week_5/vendor.protogen/
+/week_4/clean_redis/bin/
+/week_5/clean_kafka/bin/
 /week_7/grpc_with_logs/bin/
 /week_7/grpc_with_logs/vendor.protogen/
 /week_7/grpc_with_logs/logs/

week_5/clean_kafka/.env

@@ -0,0 +1,9 @@
+POSTGRES_DB=note
+POSTGRES_USER=note-user
+POSTGRES_PASSWORD=note-password
+MIGRATION_DIR=./migrations
+
+PG_DSN="host=localhost port=54321 dbname=note user=note-user password=note-password sslmode=disable"
+
+KAFKA_BROKERS=localhost:9092, localhost:9093, localhost:9094
+KAFKA_GROUP_ID=note

week_5/clean_kafka/Makefile

@@ -0,0 +1,14 @@
+include .env
+LOCAL_BIN:=$(CURDIR)/bin
+
+install-deps:
+	GOBIN=$(LOCAL_BIN) go install github.com/pressly/goose/v3/cmd/goose@v3.14.0
+
+local-migration-status:
+	$(LOCAL_BIN)/goose -dir ${MIGRATION_DIR} postgres ${PG_DSN} status -v
+
+local-migration-up:
+	$(LOCAL_BIN)/goose -dir ${MIGRATION_DIR} postgres ${PG_DSN} up -v
+
+local-migration-down:
+	$(LOCAL_BIN)/goose -dir ${MIGRATION_DIR} postgres ${PG_DSN} down -v

week_5/clean_kafka/cmd/producer/main.go

@@ -0,0 +1,67 @@
+package main
+
+import (
+	"encoding/json"
+	"log"
+	"strings"
+
+	"github.com/IBM/sarama"
+	"github.com/brianvoe/gofakeit/v6"
+
+	"github.com/olezhek28/microservices_course/week_5/clean_kafka/internal/model"
+)
+
+const (
+	brokerAddress = "localhost:9092, localhost:9093, localhost:9094"
+	topicName     = "test-topic"
+)
+
+func main() {
+	producer, err := newSyncProducer(strings.Split(brokerAddress, ","))
+	if err != nil {
+		log.Fatalf("failed to start producer: %v\n", err.Error())
+	}
+
+	defer func() {
+		if err = producer.Close(); err != nil {
+			log.Fatalf("failed to close producer: %v\n", err.Error())
+		}
+	}()
+
+	info := model.NoteInfo{
+		Title:   gofakeit.BookTitle(),
+		Content: gofakeit.Paragraph(3, 7, 5, " "),
+	}
+
+	data, err := json.Marshal(info)
+	if err != nil {
+		log.Fatalf("failed to marshal data: %v\n", err.Error())
+	}
+
+	msg := &sarama.ProducerMessage{
+		Topic: topicName,
+		Value: sarama.StringEncoder(data),
+	}
+
+	partition, offset, err := producer.SendMessage(msg)
+	if err != nil {
+		log.Printf("failed to send message in Kafka: %v\n", err.Error())
+		return
+	}
+
+	log.Printf("message sent to partition %d with offset %d\n", partition, offset)
+}
+
+func newSyncProducer(brokerList []string) (sarama.SyncProducer, error) {
+	config := sarama.NewConfig()
+	config.Producer.RequiredAcks = sarama.WaitForAll
+	config.Producer.Retry.Max = 5
+	config.Producer.Return.Successes = true
+
+	producer, err := sarama.NewSyncProducer(brokerList, config)
+	if err != nil {
+		return nil, err
+	}
+
+	return producer, nil
+}

week_5/clean_kafka/cmd/worker/main.go

@@ -0,0 +1,22 @@
+package main
+
+import (
+	"context"
+	"log"
+
+	"github.com/olezhek28/microservices_course/week_5/clean_kafka/internal/app"
+)
+
+func main() {
+	ctx := context.Background()
+
+	a, err := app.NewApp(ctx)
+	if err != nil {
+		log.Fatalf("failed to init app: %s", err.Error())
+	}
+
+	err = a.Run(ctx)
+	if err != nil {
+		log.Fatalf("failed to run app: %s", err.Error())
+	}
+}

week_5/clean_kafka/docker-compose.yaml

@@ -0,0 +1,86 @@
+version: '3.8'
+
+volumes:
+  postgres_volume:
+
+services:
+  zookeeper:
+    image: confluentinc/cp-zookeeper:7.6.1
+    hostname: zookeeper
+    container_name: zookeeper
+    ports:
+      - "2181:2181" # порт для клиентских соединений Zookeeper
+    environment:
+      ZOOKEEPER_CLIENT_PORT: 2181 # порт для клиентских соединений Zookeeper
+      ZOOKEEPER_TICK_TIME: 2000 # основной интервал времени (в миллисекундах), используемый Zookeeper
+
+  kafka1:
+    image: confluentinc/cp-kafka:7.6.1
+    hostname: kafka1
+    container_name: kafka1
+    depends_on:
+      - zookeeper # указывает, что Kafka зависит от сервиса Zookeeper и будет запущен после него
+    ports:
+      - "9092:9092" # порт для клиентских соединений Kafka
+    environment:
+      KAFKA_BROKER_ID: 1 # уникальный идентификатор брокера Kafka
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181' # адрес подключения к Zookeeper
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT # указывает протоколы безопасности для слушателей
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka1:29092,PLAINTEXT_HOST://localhost:9092 # указывает, какие адреса будут использоваться для общения с Kafka
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 # фактор репликации топика
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 # время задержки перед началом перебалансировки группы
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 # минимальное количество реплик, которые должны быть в синхронизации для топика состояния транзакции
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 # фактор репликации топика состояния транзакции
+
+  kafka2:
+    image: confluentinc/cp-kafka:7.6.1
+    hostname: kafka2
+    container_name: kafka2
+    depends_on:
+      - zookeeper # указывает, что Kafka зависит от сервиса Zookeeper и будет запущен после него
+    ports:
+      - "9093:9093" # порт для клиентских соединений Kafka
+    environment:
+      KAFKA_BROKER_ID: 2 # уникальный идентификатор брокера Kafka
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181' # адрес подключения к Zookeeper
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT # указывает протоколы безопасности для слушателей
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka2:29093,PLAINTEXT_HOST://localhost:9093 # указывает, какие адреса будут использоваться для общения с Kafka
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 # фактор репликации топика смещений
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 # время задержки перед началом перебалансировки группы
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 # минимальное количество реплик, которые должны быть в синхронизации для топика состояния транзакции
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 # фактор репликации топика состояния транзакции
+
+  kafka3:
+    image: confluentinc/cp-kafka:7.6.1
+    hostname: kafka3
+    container_name: kafka3
+    depends_on:
+      - zookeeper # указывает, что Kafka зависит от сервиса Zookeeper и будет запущен после него
+    ports:
+      - "9094:9094" # порт для клиентских соединений Kafka
+    environment:
+      KAFKA_BROKER_ID: 3 # уникальный идентификатор брокера Kafka
+      KAFKA_ZOOKEEPER_CONNECT: 'zookeeper:2181' # адрес подключения к Zookeeper
+      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT # указывает протоколы безопасности для слушателей
+      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka3:29094,PLAINTEXT_HOST://localhost:9094 # указывает, какие адреса будут использоваться для общения с Kafka
+      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 3 # фактор репликации топика смещений
+      KAFKA_GROUP_INITIAL_REBALANCE_DELAY_MS: 0 # время задержки перед началом перебалансировки группы
+      KAFKA_TRANSACTION_STATE_LOG_MIN_ISR: 2 # минимальное количество реплик, которые должны быть в синхронизации для топика состояния транзакции
+      KAFKA_TRANSACTION_STATE_LOG_REPLICATION_FACTOR: 3 # фактор репликации топика состояния транзакции
+
+  kafka-ui:
+    container_name: kafka-ui
+    image: provectuslabs/kafka-ui:v0.7.2
+    ports:
+      - "8082:8080" # порт для клиентских соединений Kafka UI
+    environment:
+      DYNAMIC_CONFIG_ENABLED: true
+
+  pg:
+    image: postgres:14-alpine3.17
+    env_file:
+      - .env
+    ports:
+      - "54321:5432"
+    volumes:
+      - postgres_volume:/var/lib/postgresql/data

week_5/clean_kafka/go.mod

@@ -0,0 +1,46 @@
+module github.com/olezhek28/microservices_course/week_5/clean_kafka
+
+go 1.22.4
+
+require (
+	github.com/IBM/sarama v1.43.2
+	github.com/Masterminds/squirrel v1.5.4
+	github.com/brianvoe/gofakeit/v6 v6.23.1
+	github.com/jackc/pgx/v4 v4.18.2
+	github.com/joho/godotenv v1.5.1
+	github.com/olezhek28/platform_common v0.0.0-20230822195735-04af626dd264
+	github.com/pkg/errors v0.9.1
+)
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/eapache/go-resiliency v1.6.0 // indirect
+	github.com/eapache/go-xerial-snappy v0.0.0-20230731223053-c322873962e3 // indirect
+	github.com/eapache/queue v1.1.0 // indirect
+	github.com/georgysavva/scany v1.2.1 // indirect
+	github.com/golang/snappy v0.0.4 // indirect
+	github.com/hashicorp/errwrap v1.0.0 // indirect
+	github.com/hashicorp/go-multierror v1.1.1 // indirect
+	github.com/hashicorp/go-uuid v1.0.3 // indirect
+	github.com/jackc/chunkreader/v2 v2.0.1 // indirect
+	github.com/jackc/pgconn v1.14.3 // indirect
+	github.com/jackc/pgio v1.0.0 // indirect
+	github.com/jackc/pgpassfile v1.0.0 // indirect
+	github.com/jackc/pgproto3/v2 v2.3.3 // indirect
+	github.com/jackc/pgservicefile v0.0.0-20221227161230-091c0ba34f0a // indirect
+	github.com/jackc/pgtype v1.14.0 // indirect
+	github.com/jackc/puddle v1.3.0 // indirect
+	github.com/jcmturner/aescts/v2 v2.0.0 // indirect
+	github.com/jcmturner/dnsutils/v2 v2.0.0 // indirect
+	github.com/jcmturner/gofork v1.7.6 // indirect
+	github.com/jcmturner/gokrb5/v8 v8.4.4 // indirect
+	github.com/jcmturner/rpc/v2 v2.0.3 // indirect
+	github.com/klauspost/compress v1.17.8 // indirect
+	github.com/lann/builder v0.0.0-20180802200727-47ae307949d0 // indirect
+	github.com/lann/ps v0.0.0-20150810152359-62de8c46ede0 // indirect
+	github.com/pierrec/lz4/v4 v4.1.21 // indirect
+	github.com/rcrowley/go-metrics v0.0.0-20201227073835-cf1acfcdf475 // indirect
+	golang.org/x/crypto v0.24.0 // indirect
+	golang.org/x/net v0.24.0 // indirect
+	golang.org/x/text v0.16.0 // indirect
+)