add concurrent_write for kafka-client

Author: bigwhite

kafka-clients/zapkafka/Makefile

@@ -0,0 +1,2 @@
+all:
+	go build github.com/bigwhite/zapkafka/cmd/concurrent_write

kafka-clients/zapkafka/cmd/concurrent_write/main.go

@@ -0,0 +1,46 @@
+package main
+
+import (
+	"fmt"
+	"os"
+	"sync"
+	"sync/atomic"
+	"time"
+
+	log "github.com/bigwhite/zapkafka"
+	"go.uber.org/zap"
+	"go.uber.org/zap/zapcore"
+)
+
+func SaramaProducer() {
+	p, err := log.NewKafkaAsyncProducer([]string{"localhost:29092"})
+	if err != nil {
+		panic(err)
+	}
+	logger := log.New(log.NewKafkaSyncer(p, "test", zapcore.AddSync(os.Stderr)), int8(0))
+	var wg sync.WaitGroup
+	var cnt int64
+
+	for j := 0; j < 10; j++ {
+		wg.Add(1)
+		go func(j int) {
+			var value string
+			for i := 0; i < 10000; i++ {
+				now := time.Now()
+				value = fmt.Sprintf("%02d-%04d-%s", j, i, now.Format("15:04:05"))
+				logger.Info("log message:", zap.String("value", value))
+				atomic.AddInt64(&cnt, 1)
+			}
+			wg.Done()
+		}(j)
+	}
+
+	wg.Wait()
+	logger.Sync()
+	println("cnt =", atomic.LoadInt64(&cnt))
+	time.Sleep(10 * time.Second)
+}
+
+func main() {
+	SaramaProducer()
+}

kafka-clients/zapkafka/kafka_syncer.go

@@ -44,13 +44,16 @@ func (ws *kafkaWriteSyncer) Write(b []byte) (n int, err error) {
 		Topic: ws.topic,
 		Value: sarama.ByteEncoder(b1),
 	}
+	ws.producer.Input() <- msg
 
-	select {
-	case ws.producer.Input() <- msg:
-	default:
-		// if producer block on input channel, write log entry to default fallbackSyncer
-		return ws.fallbackSyncer.Write(b1)
-	}
+	/*
+		select {
+		case ws.producer.Input() <- msg:
+		default:
+			// if producer block on input channel, write log entry to default fallbackSyncer
+			return ws.fallbackSyncer.Write(b1)
+		}
+	*/
 
 	return len(b1), nil
 }

kafka-clients/zapkafka/log.go

@@ -18,6 +18,10 @@ func (l *Logger) Info(msg string, fields ...zap.Field) {
 	l.l.Info(msg, fields...)
 }
 
+func (l *Logger) Sync() {
+	l.l.Sync()
+}
+
 func New(writer io.Writer, level int8, opts ...zap.Option) *Logger {
 	if writer == nil {
 		panic("the writer is nil")

kafka-clients/zapkafka/log_test.go

@@ -13,6 +13,7 @@ import (
 	"go.uber.org/zap"
 )
 
+/*
 func TestWriteFailWithKafkaSyncer(t *testing.T) {
 	config := sarama.NewConfig()
 	config.Producer.Return.Successes = true
@@ -46,6 +47,30 @@ func TestWriteFailWithKafkaSyncer(t *testing.T) {
 		t.Errorf("want true, actual false")
 	}
 }
+*/
+
+func TestWriteFailWithKafkaSyncer(t *testing.T) {
+	config := sarama.NewConfig()
+	p := mocks.NewAsyncProducer(t, config)
+
+	var buf = []byte{}
+	w := bytes.NewBuffer(buf)
+	logger := New(NewKafkaSyncer(p, "test", NewFileSyncer(w)), 0)
+
+	// all below will be written to the fallback sycner
+	logger.Info("demo1", zap.String("status", "ok")) // write to the kafka syncer
+
+	p.ExpectInputAndFail(errors.New("produce error"))
+
+	s := string(w.Bytes())
+	if !strings.Contains(s, "demo1") {
+		t.Errorf("want true, actual false")
+	}
+
+	if err := p.Close(); err != nil {
+		t.Error(err)
+	}
+}
 
 func TestWriteOKWithKafkaSyncer(t *testing.T) {
 	config := sarama.NewConfig()