OOM

[chushuai]

I have 7 billion data to be written to kafka, and when I write 200 million, the memory OOM, The following is part of my code logic, which can help you see what the problem is?

       producer, err :=kafka.NewProducer(&kafka.ConfigMap{"bootstrap.servers": bootstrapServers})
       domainList := []string{}
	scanner := bufio.NewScanner(inFIle)
	count := 0
	for scanner.Scan() {
		lineText := strings.TrimSpace(scanner.Text())
		if lineText == "" {
			continue
		}
		count += 1
		if count%1000000 == 0 {
			logger.Infof("Already read %d lines", count)
		}
		domainList = append(domainList, lineText)
		if len(domainList) >= 1000 {
			raw, _ := json.Marshal(domainList)
			err = producer.Produce(&kafka.Message{
				TopicPartition: kafka.TopicPartition{Topic: &topic, Partition: kafka.PartitionAny},
				Value:          raw,
			}, nil)
			if err != nil {
				logger.Error(err)
				if err.(kafka.Error).Code() == kafka.ErrQueueFull {
					logger.Info("kafka queue is full, wait 1s for messages to be delivered then try again")
					time.Sleep(10*time.Second)
					continue
				}
			}
			domainList = []string{}
		}
	}

[milindl]

Just a question, are you reading from the producer.Events() anywhere (in a different goroutine perhaps)? Otherwise, the channel will grow for every message that is delivered.

[chushuai]

I did not read any messages from producer.Events() anywhere, nor did the example code contain any references to producer.Events() @milindl

[milindl]

You can take a look at this example: https://github.com/confluentinc/confluent-kafka-go/blob/master/examples/producer_example/producer_example.go#L51
You should read the messages from producer.Events(), as well, in your code.