Golang ve C# ile Apache Kafka Uygulamaları: Karşılaştırmalı Bir İnceleme

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78

package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"time"

	"github.com/Shopify/sarama"
	"github.com/go-sql-driver/mysql"
)

// MySQL'den veri çeken ve Kafka'ya gönderen basit bir connector örneği
func main() {
	// MySQL bağlantısı
	db, err := sql.Open("mysql", "user:password@tcp(127.0.0.1:3306)/database")
	if err != nil {
		log.Fatal(err)
	}
	defer db.Close()

	// Kafka producer yapılandırması
	config := sarama.NewConfig()
	config.Producer.Return.Successes = true
	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, config)
	if err != nil {
		log.Fatal(err)
	}
	defer producer.Close()

	// Periyodik olarak veri çekme ve gönderme
	ticker := time.NewTicker(1 * time.Minute)
	defer ticker.Stop()

	for range ticker.C {
		// Veritabanından veri çekme
		rows, err := db.Query("SELECT id, name, created_at FROM users WHERE updated_at > ?", 
			time.Now().Add(-1*time.Minute))
		if err != nil {
			log.Println("Query error:", err)
			continue
		}

		for rows.Next() {
			var id int
			var name string
			var createdAt time.Time
			
			if err := rows.Scan(&id, &name, &createdAt); err != nil {
				log.Println("Scan error:", err)
				continue
			}
			
			// Veriyi JSON formatına dönüştürme
			data, err := json.Marshal(map[string]interface{}{
				"id": id,
				"name": name,
				"created_at": createdAt,
			})
			if err != nil {
				log.Println("JSON error:", err)
				continue
			}
			
			// Kafka'ya gönderme
			_, _, err = producer.SendMessage(&sarama.ProducerMessage{
				Topic: "db-changes",
				Key:   sarama.StringEncoder(fmt.Sprintf("%d", id)),
				Value: sarama.ByteEncoder(data),
			})
			if err != nil {
				log.Println("Send error:", err)
			}
		}
		rows.Close()
	}
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92

using System;
using System.Data.SqlClient;
using System.Text.Json;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace SqlServerKafkaConnector
{
    public class SqlServerSourceConnector : BackgroundService
    {
        private readonly ILogger<SqlServerSourceConnector> _logger;
        private readonly string _connectionString;
        private readonly string _topic;
        private readonly string _bootstrapServers;
        private DateTime _lastCheckpoint = DateTime.MinValue;

        public SqlServerSourceConnector(
            ILogger<SqlServerSourceConnector> logger,
            string connectionString,
            string topic,
            string bootstrapServers)
        {
            _logger = logger;
            _connectionString = connectionString;
            _topic = topic;
            _bootstrapServers = bootstrapServers;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            var config = new ProducerConfig { BootstrapServers = _bootstrapServers };
            
            using var producer = new ProducerBuilder<string, string>(config).Build();
            
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    await PollDatabaseChanges(producer);
                    await Task.Delay(TimeSpan.FromMinutes(1), stoppingToken);
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error polling database changes");
                    await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
                }
            }
        }

        private async Task PollDatabaseChanges(IProducer<string, string> producer)
        {
            using var connection = new SqlConnection(_connectionString);
            await connection.OpenAsync();
            
            var query = @"SELECT Id, Name, CreatedAt FROM Users WHERE UpdatedAt > @LastCheckpoint";
            
            using var command = new SqlCommand(query, connection);
            command.Parameters.AddWithValue("@LastCheckpoint", _lastCheckpoint);
            
            using var reader = await command.ExecuteReaderAsync();
            
            while (await reader.ReadAsync())
            {
                var id = reader.GetInt32(0);
                var name = reader.GetString(1);
                var createdAt = reader.GetDateTime(2);
                
                var data = new
                {
                    Id = id,
                    Name = name,
                    CreatedAt = createdAt
                };
                
                var json = JsonSerializer.Serialize(data);
                
                await producer.ProduceAsync(_topic, new Message<string, string>
                {
                    Key = id.ToString(),
                    Value = json
                });
                
                _logger.LogInformation("Sent record: {Id}", id);
            }
            
            _lastCheckpoint = DateTime.UtcNow;
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

package main

import (
	"context"
	"encoding/json"
	"log"
	"strings"
	"sync"
	"time"

	"github.com/Shopify/sarama"
)

// Basit bir stream processing uygulaması
func main() {
	// Consumer yapılandırması
	consumerConfig := sarama.NewConfig()
	consumerConfig.Consumer.Return.Errors = true
	consumerConfig.Consumer.Offsets.Initial = sarama.OffsetOldest

	// Producer yapılandırması
	producerConfig := sarama.NewConfig()
	producerConfig.Producer.Return.Successes = true

	// Consumer grup oluşturma
	consumer, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "stream-processor", consumerConfig)
	if err != nil {
		log.Fatalf("Error creating consumer group: %v", err)
	}
	defer consumer.Close()

	// Producer oluşturma
	producer, err := sarama.NewSyncProducer([]string{"localhost:9092"}, producerConfig)
	if err != nil {
		log.Fatalf("Error creating producer: %v", err)
	}
	defer producer.Close()

	// Stream processor
	handler := &StreamProcessor{
		producer: producer,
		// 5 dakikalık bir pencere için basit bir in-memory state
		windowedCounts: make(map[string]int),
		windowStart:    time.Now(),
		windowSize:     5 * time.Minute,
	}

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Tüketim döngüsü
	topics := []string{"input-topic"}
	for {
		if err := consumer.Consume(ctx, topics, handler); err != nil {
			log.Fatalf("Error from consumer: %v", err)
		}
		
		if ctx.Err() != nil {
			return
		}
	}
}

// StreamProcessor, sarama.ConsumerGroupHandler arayüzünü uygular
type StreamProcessor struct {
	producer       sarama.SyncProducer
	windowedCounts map[string]int
	windowStart    time.Time
	windowSize     time.Duration
	mu             sync.Mutex
}

func (p *StreamProcessor) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (p *StreamProcessor) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (p *StreamProcessor) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for message := range claim.Messages() {
		// Mesajı işle
		var data map[string]interface{}
		if err := json.Unmarshal(message.Value, &data); err != nil {
			log.Printf("Error unmarshaling message: %v", err)
			session.MarkMessage(message, "")
			continue
		}
		
		// Örnek işlem: Kelime sayma
		if text, ok := data["text"].(string); ok {
			words := strings.Fields(text)
			
			p.mu.Lock()
			// Pencere süresi dolduysa, sonuçları gönder ve pencereyi sıfırla
			if time.Since(p.windowStart) > p.windowSize {
				for word, count := range p.windowedCounts {
					result, _ := json.Marshal(map[string]interface{}{
						"word":       word,
						"count":      count,
						"windowEnd":  time.Now(),
						"windowStart": p.windowStart,
					})
					
					p.producer.SendMessage(&sarama.ProducerMessage{
						Topic: "word-counts",
						Key:   sarama.StringEncoder(word),
						Value: sarama.ByteEncoder(result),
					})
				}
				
				p.windowedCounts = make(map[string]int)
				p.windowStart = time.Now()
			}
			
			// Kelimeleri say
			for _, word := range words {
				word = strings.ToLower(strings.Trim(word, ".,!?\"':;()"))
				if word != "" {
					p.windowedCounts[word]++
				}
			}
			p.mu.Unlock()
		}
		
		session.MarkMessage(message, "")
	}
	return nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149

using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.Linq;
using System.Text.Json;
using System.Text.RegularExpressions;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;

namespace KafkaStreamProcessor
{
    public class WordCountProcessor : BackgroundService
    {
        private readonly ILogger<WordCountProcessor> _logger;
        private readonly string _bootstrapServers;
        private readonly string _inputTopic;
        private readonly string _outputTopic;
        private readonly TimeSpan _windowSize = TimeSpan.FromMinutes(5);
        
        private ConcurrentDictionary<string, int> _wordCounts = new ConcurrentDictionary<string, int>();
        private DateTime _windowStart = DateTime.UtcNow;

        public WordCountProcessor(
            ILogger<WordCountProcessor> logger,
            string bootstrapServers,
            string inputTopic,
            string outputTopic)
        {
            _logger = logger;
            _bootstrapServers = bootstrapServers;
            _inputTopic = inputTopic;
            _outputTopic = outputTopic;
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            var consumerConfig = new ConsumerConfig
            {
                BootstrapServers = _bootstrapServers,
                GroupId = "word-count-processor",
                AutoOffsetReset = AutoOffsetReset.Earliest,
                EnableAutoCommit = false
            };

            var producerConfig = new ProducerConfig
            {
                BootstrapServers = _bootstrapServers
            };

            using var consumer = new ConsumerBuilder<Ignore, string>(consumerConfig).Build();
            using var producer = new ProducerBuilder<string, string>(producerConfig).Build();
            
            consumer.Subscribe(_inputTopic);
            
            try
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    try
                    {
                        var consumeResult = consumer.Consume(stoppingToken);
                        
                        // Mesajı işle
                        try
                        {
                            var data = JsonSerializer.Deserialize<Dictionary<string, JsonElement>>(consumeResult.Message.Value);
                            
                            if (data.TryGetValue("text", out var textElement) && textElement.ValueKind == JsonValueKind.String)
                            {
                                var text = textElement.GetString();
                                if (!string.IsNullOrEmpty(text))
                                {
                                    ProcessText(text);
                                }
                            }
                        }
                        catch (JsonException ex)
                        {
                            _logger.LogError(ex, "Error deserializing message");
                        }
                        
                        // Pencere süresi dolduysa, sonuçları gönder
                        if (DateTime.UtcNow - _windowStart > _windowSize)
                        {
                            await SendWindowResults(producer);
                        }
                        
                        consumer.Commit(consumeResult);
                    }
                    catch (ConsumeException ex)
                    {
                        _logger.LogError(ex, "Error consuming message");
                    }
                }
            }
            finally
            {
                consumer.Close();
            }
        }

        private void ProcessText(string text)
        {
            var words = Regex.Split(text, @"\W+")
                .Where(w => !string.IsNullOrEmpty(w))
                .Select(w => w.ToLowerInvariant());
            
            foreach (var word in words)
            {
                _wordCounts.AddOrUpdate(word, 1, (_, count) => count + 1);
            }
        }

        private async Task SendWindowResults(IProducer<string, string> producer)
        {
            var windowEnd = DateTime.UtcNow;
            var tasks = new List<Task>();
            
            foreach (var entry in _wordCounts.ToArray())
            {
                var result = new
                {
                    Word = entry.Key,
                    Count = entry.Value,
                    WindowStart = _windowStart,
                    WindowEnd = windowEnd
                };
                
                var json = JsonSerializer.Serialize(result);
                
                tasks.Add(producer.ProduceAsync(_outputTopic, new Message<string, string>
                {
                    Key = entry.Key,
                    Value = json
                }));
                
                _logger.LogInformation("Word count: {Word} = {Count}", entry.Key, entry.Value);
            }
            
            await Task.WhenAll(tasks);
            
            _wordCounts.Clear();
            _windowStart = DateTime.UtcNow;
        }
    }
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74

package main

import (
	"crypto/tls"
	"crypto/x509"
	"io/ioutil"
	"log"

	"github.com/Shopify/sarama"
)

func main() {
	// TLS yapılandırması
	tlsConfig, err := createTLSConfig("client.cer.pem", "client.key.pem", "ca.pem")
	if err != nil {
		log.Fatalf("Error creating TLS config: %v", err)
	}

	// Kafka yapılandırması
	config := sarama.NewConfig()
	config.Net.TLS.Enable = true
	config.Net.TLS.Config = tlsConfig
	config.Net.SASL.Enable = true
	config.Net.SASL.Mechanism = sarama.SASLTypePlaintext
	config.Net.SASL.User = "username"
	config.Net.SASL.Password = "password"
	config.Producer.Return.Successes = true

	// Producer oluşturma
	producer, err := sarama.NewSyncProducer([]string{"kafka:9093"}, config)
	if err != nil {
		log.Fatalf("Error creating producer: %v", err)
	}
	defer producer.Close()

	// Mesaj gönderme
	msg := &sarama.ProducerMessage{
		Topic: "secure-topic",
		Value: sarama.StringEncoder("Secure message"),
	}

	partition, offset, err := producer.SendMessage(msg)
	if err != nil {
		log.Fatalf("Error sending message: %v", err)
	}

	log.Printf("Message sent to partition %d at offset %d", partition, offset)
}

func createTLSConfig(clientCertFile, clientKeyFile, caCertFile string) (*tls.Config, error) {
	// CA sertifikasını yükle
	caCert, err := ioutil.ReadFile(caCertFile)
	if err != nil {
		return nil, err
	}

	caCertPool := x509.NewCertPool()
	caCertPool.AppendCertsFromPEM(caCert)

	// İstemci sertifikası ve anahtarını yükle
	cert, err := tls.LoadX509KeyPair(clientCertFile, clientKeyFile)
	if err != nil {
		return nil, err
	}

	// TLS yapılandırması oluştur
	tlsConfig := &tls.Config{
		Certificates:       []tls.Certificate{cert},
		RootCAs:            caCertPool,
		InsecureSkipVerify: false,
	}

	return tlsConfig, nil
}

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45

using System;
using System.IO;
using System.Security.Cryptography.X509Certificates;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;

namespace SecureKafkaClient
{
    class Program
    {
        static async Task Main(string[] args)
        {
            var config = new ProducerConfig
            {
                BootstrapServers = "kafka:9093",
                
                // SSL/TLS yapılandırması
                SecurityProtocol = SecurityProtocol.SaslSsl,
                SslCaLocation = "ca.pem",
                SslCertificateLocation = "client.cer.pem",
                SslKeyLocation = "client.key.pem",
                
                // SASL yapılandırması
                SaslMechanism = SaslMechanism.Plain,
                SaslUsername = "username",
                SaslPassword = "password"
            };

            using var producer = new ProducerBuilder<Null, string>(config).Build();
            
            try
            {
                var result = await producer.ProduceAsync("secure-topic", 
                    new Message<Null, string> { Value = "Secure message" });
                
                Console.WriteLine($"Message sent to {result.TopicPartitionOffset}");
            }
            catch (ProduceException<Null, string> e)
            {
                Console.WriteLine($"Error sending message: {e.Error.Reason}");
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116

package main

import (
	"log"
	"net/http"
	"time"

	"github.com/Shopify/sarama"
	"github.com/prometheus/client_golang/prometheus"
	"github.com/prometheus/client_golang/prometheus/promauto"
	"github.com/prometheus/client_golang/prometheus/promhttp"
)

var (
	// Prometheus metrikleri
	messagesProcessed = promauto.NewCounter(prometheus.CounterOpts{
		Name: "kafka_consumer_messages_processed_total",
		Help: "The total number of processed messages",
	})
	
	processingLatency = promauto.NewHistogram(prometheus.HistogramOpts{
		Name:    "kafka_consumer_processing_latency_seconds",
		Help:    "Message processing latency in seconds",
		Buckets: prometheus.DefBuckets,
	})
	
	consumerLag = promauto.NewGaugeVec(prometheus.GaugeOpts{
		Name: "kafka_consumer_lag",
		Help: "The lag of the consumer per partition",
	}, []string{"topic", "partition"})
)

func main() {
	// Prometheus HTTP sunucusu
	go func() {
		http.Handle("/metrics", promhttp.Handler())
		log.Fatal(http.ListenAndServe(":2112", nil))
	}()

	// Kafka consumer yapılandırması
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.Initial = sarama.OffsetOldest

	consumer, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "monitored-consumer", config)
	if err != nil {
		log.Fatalf("Error creating consumer group: %v", err)
	}
	defer consumer.Close()

	// Consumer handler
	handler := &MonitoredConsumerHandler{}
	
	// Tüketim döngüsü
	topics := []string{"monitored-topic"}
	for {
		if err := consumer.Consume(context.Background(), topics, handler); err != nil {
			log.Fatalf("Error from consumer: %v", err)
		}
	}
}

// MonitoredConsumerHandler, sarama.ConsumerGroupHandler arayüzünü uygular
type MonitoredConsumerHandler struct{}

func (h *MonitoredConsumerHandler) Setup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *MonitoredConsumerHandler) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *MonitoredConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	// Tüketici gecikmesini izle
	go monitorConsumerLag(session, claim)
	
	for message := range claim.Messages() {
		// İşleme başlangıç zamanı
		startTime := time.Now()
		
		// Mesajı işle
		processMessage(message)
		
		// Metrikleri güncelle
		messagesProcessed.Inc()
		processingLatency.Observe(time.Since(startTime).Seconds())
		
		// Mesajı işaretleme
		session.MarkMessage(message, "")
	}
	return nil
}

func monitorConsumerLag(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) {
	ticker := time.NewTicker(10 * time.Second)
	defer ticker.Stop()
	
	topic := claim.Topic()
	partition := claim.Partition()
	
	for range ticker.C {
		// Tüketici gecikmesini hesapla
		currentOffset := session.MarkOffset(topic, partition, 0, "")
		highWatermark := claim.HighWaterMarkOffset()
		lag := highWatermark - currentOffset
		
		// Metriği güncelle
		consumerLag.WithLabelValues(topic, fmt.Sprintf("%d", partition)).Set(float64(lag))
	}
}

func processMessage(message *sarama.ConsumerMessage) {
	// Mesaj işleme mantığı
	time.Sleep(100 * time.Millisecond) // Simüle edilmiş işleme süresi
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130

using System;
using System.Collections.Generic;
using System.Diagnostics;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Prometheus;

namespace KafkaMonitoringExample
{
    public class MonitoredKafkaConsumer : BackgroundService
    {
        private readonly ILogger<MonitoredKafkaConsumer> _logger;
        private readonly ConsumerConfig _config;
        private readonly string _topic;
        
        // Prometheus metrikleri
        private static readonly Counter MessagesProcessed = Metrics.CreateCounter(
            "kafka_consumer_messages_processed_total", 
            "The total number of processed messages");
        
        private static readonly Histogram ProcessingLatency = Metrics.CreateHistogram(
            "kafka_consumer_processing_latency_seconds",
            "Message processing latency in seconds",
            new HistogramConfiguration
            {
                Buckets = Histogram.ExponentialBuckets(0.001, 2, 10)
            });
        
        private static readonly Gauge ConsumerLag = Metrics.CreateGauge(
            "kafka_consumer_lag",
            "The lag of the consumer per partition",
            new GaugeConfiguration
            {
                LabelNames = new[] { "topic", "partition" }
            });

        public MonitoredKafkaConsumer(
            ILogger<MonitoredKafkaConsumer> logger,
            ConsumerConfig config,
            string topic)
        {
            _logger = logger;
            _config = config;
            _topic = topic;
            
            // Prometheus HTTP sunucusu başlat
            var server = new MetricServer(port: 9090);
            server.Start();
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            using var consumer = new ConsumerBuilder<Ignore, string>(_config).Build();
            consumer.Subscribe(_topic);
            
            // Tüketici gecikmesini izleme görevi
            _ = Task.Run(() => MonitorConsumerLag(consumer, stoppingToken), stoppingToken);
            
            try
            {
                while (!stoppingToken.IsCancellationRequested)
                {
                    try
                    {
                        var consumeResult = consumer.Consume(stoppingToken);
                        
                        // İşleme başlangıç zamanı
                        var stopwatch = Stopwatch.StartNew();
                        
                        // Mesajı işle
                        await ProcessMessageAsync(consumeResult.Message.Value);
                        
                        // Metrikleri güncelle
                        MessagesProcessed.Inc();
                        ProcessingLatency.Observe(stopwatch.Elapsed.TotalSeconds);
                        
                        // Mesajı işaretleme
                        consumer.Commit(consumeResult);
                    }
                    catch (ConsumeException ex)
                    {
                        _logger.LogError(ex, "Error consuming message");
                    }
                }
            }
            finally
            {
                consumer.Close();
            }
        }

        private async Task ProcessMessageAsync(string message)
        {
            // Mesaj işleme mantığı
            await Task.Delay(100); // Simüle edilmiş işleme süresi
        }
        
        private async Task MonitorConsumerLag(IConsumer<Ignore, string> consumer, CancellationToken stoppingToken)
        {
            while (!stoppingToken.IsCancellationRequested)
            {
                try
                {
                    var watermarkOffsets = consumer.GetWatermarkOffsets(
                        new TopicPartition(_topic, new Partition(0)));
                    
                    var committedOffsets = consumer.Committed(
                        new List<TopicPartition> { new TopicPartition(_topic, new Partition(0)) },
                        TimeSpan.FromSeconds(10));
                    
                    if (committedOffsets.Count > 0)
                    {
                        var lag = watermarkOffsets.High - committedOffsets[0].Offset;
                        ConsumerLag.WithLabels(_topic, "0").Set(lag.Value);
                        _logger.LogInformation("Consumer lag: {Lag}", lag);
                    }
                }
                catch (Exception ex)
                {
                    _logger.LogError(ex, "Error monitoring consumer lag");
                }
                
                await Task.Delay(TimeSpan.FromSeconds(10), stoppingToken);
            }
        }
    }
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175

package main

import (
	"context"
	"errors"
	"log"
	"os"
	"os/signal"
	"syscall"
	"time"

	"github.com/Shopify/sarama"
	"github.com/cenkalti/backoff/v4"
)

func main() {
	// Graceful shutdown için sinyal yakalama
	signals := make(chan os.Signal, 1)
	signal.Notify(signals, syscall.SIGINT, syscall.SIGTERM)

	ctx, cancel := context.WithCancel(context.Background())
	defer cancel()

	// Sinyal yakalama goroutine
	go func() {
		<-signals
		log.Println("Shutdown signal received, closing consumer...")
		cancel()
	}()

	// Kafka consumer'ı başlat
	if err := startConsumerWithRetry(ctx); err != nil {
		log.Fatalf("Fatal error: %v", err)
	}
}

func startConsumerWithRetry(ctx context.Context) error {
	// Exponential backoff stratejisi
	expBackoff := backoff.NewExponentialBackOff()
	expBackoff.MaxElapsedTime = 5 * time.Minute

	return backoff.RetryNotify(
		func() error {
			return startConsumer(ctx)
		},
		expBackoff,
		func(err error, duration time.Duration) {
			log.Printf("Error: %v, retrying in %v...", err, duration)
		},
	)
}

func startConsumer(ctx context.Context) error {
	// Kafka yapılandırması
	config := sarama.NewConfig()
	config.Consumer.Return.Errors = true
	config.Consumer.Offsets.Initial = sarama.OffsetOldest
	config.Consumer.Offsets.AutoCommit.Enable = false
	config.Consumer.Retry.Backoff = 1 * time.Second
	config.Consumer.Retry.Max = 5

	// Consumer grup oluşturma
	group, err := sarama.NewConsumerGroup([]string{"localhost:9092"}, "resilient-consumer", config)
	if err != nil {
		return err
	}
	defer group.Close()

	// Consumer hata kanalını izleme
	go func() {
		for err := range group.Errors() {
			log.Printf("Consumer group error: %v", err)
		}
	}()

	// Consumer handler
	handler := &ResilientConsumerHandler{
		readyC: make(chan bool),
	}

	// Tüketim döngüsü
	topics := []string{"resilient-topic"}
	
	// Handler hazır olana kadar bekle
	go func() {
		if err := group.Consume(ctx, topics, handler); err != nil {
			log.Printf("Error from consumer: %v", err)
		}
	}()

	<-handler.readyC
	log.Println("Consumer is ready")

	// Bağlam iptal edilene kadar bekle
	<-ctx.Done()
	log.Println("Context cancelled, closing consumer")
	
	// Consumer'ın kapanması için biraz bekle
	time.Sleep(2 * time.Second)
	return nil
}

// ResilientConsumerHandler, sarama.ConsumerGroupHandler arayüzünü uygular
type ResilientConsumerHandler struct {
	readyC chan bool
}

func (h *ResilientConsumerHandler) Setup(session sarama.ConsumerGroupSession) error {
	// Consumer hazır olduğunda bildir
	close(h.readyC)
	return nil
}

func (h *ResilientConsumerHandler) Cleanup(session sarama.ConsumerGroupSession) error {
	return nil
}

func (h *ResilientConsumerHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error {
	for {
		select {
		case message, ok := <-claim.Messages():
			if !ok {
				return nil
			}
			
			// Mesajı işle
			if err := processMessageWithRetry(message); err != nil {
				// Kritik hata durumunda işlemi durdur
				if errors.Is(err, ErrCritical) {
					return err
				}
				
				// Geçici hata durumunda mesajı işaretleme ve devam et
				log.Printf("Error processing message, skipping: %v", err)
			}
			
			// Mesajı işaretleme
			session.MarkMessage(message, "")
			
		case <-session.Context().Done():
			return nil
		}
	}
}

// Özel hata türleri
var ErrCritical = errors.New("critical error")
var ErrTemporary = errors.New("temporary error")

func processMessageWithRetry(message *sarama.ConsumerMessage) error {
	// Retry stratejisi
	expBackoff := backoff.NewExponentialBackOff()
	expBackoff.MaxElapsedTime = 30 * time.Second

	return backoff.Retry(
		func() error {
			err := processMessage(message)
			
			// Geçici hata durumunda yeniden dene
			if errors.Is(err, ErrTemporary) {
				return err
			}
			
			// Diğer hata türlerinde yeniden deneme
			return nil
		},
		expBackoff,
	)
}

func processMessage(message *sarama.ConsumerMessage) error {
	// Mesaj işleme mantığı
	// ...
	return nil
}

  1
  2
  3
  4
  5
  6
  7
  8
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222

using System;
using System.Threading;
using System.Threading.Tasks;
using Confluent.Kafka;
using Microsoft.Extensions.Hosting;
using Microsoft.Extensions.Logging;
using Polly;
using Polly.Retry;

namespace ResilientKafkaConsumer
{
    public class ResilientKafkaConsumerService : BackgroundService
    {
        private readonly ILogger<ResilientKafkaConsumerService> _logger;
        private readonly ConsumerConfig _config;
        private readonly string _topic;
        private readonly AsyncRetryPolicy _retryPolicy;

        public ResilientKafkaConsumerService(
            ILogger<ResilientKafkaConsumerService> logger,
            ConsumerConfig config,
            string topic)
        {
            _logger = logger;
            _config = config;
            _topic = topic;
            
            // Polly retry politikası
            _retryPolicy = Policy
                .Handle<Exception>(ex => !(ex is OperationCanceledException) && IsTransientException(ex))
                .WaitAndRetryAsync(
                    5,
                    retryAttempt => TimeSpan.FromSeconds(Math.Pow(2, retryAttempt)),
                    (exception, timeSpan, retryCount, context) =>
                    {
                        _logger.LogWarning(
                            exception, 
                            "Error processing message. Retrying in {RetryTimeSpan}s. Attempt {RetryCount}",
                            timeSpan.TotalSeconds,
                            retryCount);
                    }
                );
        }

        protected override async Task ExecuteAsync(CancellationToken stoppingToken)
        {
            _logger.LogInformation("Resilient Kafka consumer service starting");

            // Graceful shutdown için CancellationToken kullanımı
            using var linkedCts = CancellationTokenSource.CreateLinkedTokenSource(stoppingToken);
            
            try
            {
                await StartConsumerWithRetryAsync(linkedCts.Token);
            }
            catch (OperationCanceledException)
            {
                _logger.LogInformation("Consumer service was cancelled");
            }
            catch (Exception ex)
            {
                _logger.LogError(ex, "Fatal error in consumer service");
                throw;
            }
            finally
            {
                _logger.LogInformation("Resilient Kafka consumer service stopping");
            }
        }

        private async Task StartConsumerWithRetryAsync(CancellationToken cancellationToken)
        {
            // Bağlantı kurma yeniden deneme politikası
            var connectionRetryPolicy = Policy
                .Handle<Exception>(ex => !(ex is OperationCanceledException))
                .WaitAndRetryForeverAsync(
                    retryAttempt => TimeSpan.FromSeconds(Math.Min(30, Math.Pow(2, retryAttempt))),
                    (exception, timeSpan, context) =>
                    {
                        _logger.LogWarning(
                            exception,
                            "Error connecting to Kafka. Retrying in {RetryTimeSpan}s",
                            timeSpan.TotalSeconds);
                    }
                );

            await connectionRetryPolicy.ExecuteAsync(async () =>
            {
                using var consumer = new ConsumerBuilder<Ignore, string>(_config)
                    .SetErrorHandler((_, error) =>
                    {
                        _logger.LogError("Kafka error: {ErrorReason}", error.Reason);
                    })
                    .SetPartitionsAssignedHandler((c, partitions) =>
                    {
                        _logger.LogInformation("Partitions assigned: {Partitions}", string.Join(", ", partitions));
                    })
                    .SetPartitionsRevokedHandler((c, partitions) =>
                    {
                        _logger.LogInformation("Partitions revoked: {Partitions}", string.Join(", ", partitions));
                    })
                    .Build();

                try
                {
                    consumer.Subscribe(_topic);
                    _logger.LogInformation("Subscribed to topic: {Topic}", _topic);

                    while (!cancellationToken.IsCancellationRequested)
                    {
                        try
                        {
                            var consumeResult = consumer.Consume(cancellationToken);
                            
                            // Mesajı işle
                            await ProcessMessageWithRetryAsync(consumeResult, cancellationToken);
                            
                            // Başarılı işleme sonrası offset'i commit et
                            try
                            {
                                consumer.Commit(consumeResult);
                            }
                            catch (KafkaException ex)
                            {
                                _logger.LogWarning(ex, "Error committing offset");
                            }
                        }
                        catch (ConsumeException ex)
                        {
                            _logger.LogError(ex, "Error consuming message");
                        }
                    }
                }
                finally
                {
                    // Graceful shutdown
                    try
                    {
                        consumer.Close();
                        _logger.LogInformation("Consumer closed");
                    }
                    catch (Exception ex)
                    {
                        _logger.LogWarning(ex, "Error closing consumer");
                    }
                }
            });
        }

        private async Task ProcessMessageWithRetryAsync(ConsumeResult<Ignore, string> consumeResult, CancellationToken cancellationToken)
        {
            await _retryPolicy.ExecuteAsync(async () =>
            {
                try
                {
                    await ProcessMessageAsync(consumeResult.Message.Value, cancellationToken);
                    _logger.LogInformation(
                        "Processed message at {TopicPartitionOffset}",
                        consumeResult.TopicPartitionOffset);
                }
                catch (Exception ex) when (IsCriticalException(ex))
                {
                    _logger.LogCritical(ex, "Critical error processing message");
                    // Kritik hataları yeniden atmak için
                    throw;
                }
                catch (Exception ex) when (!IsTransientException(ex))
                {
                    _logger.LogError(ex, "Non-transient error processing message");
                    // Geçici olmayan hataları yeniden atmak için
                    throw;
                }
            });
        }

        private async Task ProcessMessageAsync(string message, CancellationToken cancellationToken)
        {
            // Mesaj işleme mantığı
            await Task.Delay(100, cancellationToken); // Simüle edilmiş işleme
            
            // Rastgele hata simülasyonu (test amaçlı)
            if (new Random().Next(100) < 10)
            {
                throw new TransientException("Simulated transient error");
            }
        }

        private bool IsTransientException(Exception ex)
        {
            // Geçici hataları tanımlama
            return ex is TransientException ||
                   ex is KafkaException kafkaEx && IsTransientKafkaError(kafkaEx.Error.Code);
        }

        private bool IsCriticalException(Exception ex)
        {
            // Kritik hataları tanımlama
            return ex is CriticalException ||
                   ex is OutOfMemoryException;
        }

        private bool IsTransientKafkaError(ErrorCode errorCode)
        {
            // Geçici Kafka hatalarını tanımlama
            return errorCode == ErrorCode.NetworkException ||
                   errorCode == ErrorCode.AllBrokersDown ||
                   errorCode == ErrorCode.NotCoordinatorForGroup ||
                   errorCode == ErrorCode.RequestTimedOut;
        }
    }

    // Özel hata türleri
    public class TransientException : Exception
    {
        public TransientException(string message) : base(message) { }
    }

    public class CriticalException : Exception
    {
        public CriticalException(string message) : base(message) { }
    }
}