Function consumeMsgs

functional_consumer_test.go:607–668

(t *testing.T, clientVersions []KafkaVersion, producedMessages []*ProducerMessage)

}

func consumeMsgs(t *testing.T, clientVersions []KafkaVersion, producedMessages []*ProducerMessage) {
	// Consume all produced messages with all client versions supported by the
	// cluster.
	g := errgroup.Group{}
	for _, consVer := range clientVersions {
		// Create a partition consumer that should start from the first produced
		// message.
		consCfg := NewFunctionalTestConfig()
		consCfg.ClientID = t.Name() + "-Consumer-" + consVer.String()
		consCfg.Consumer.MaxProcessingTime = time.Second
		consCfg.Metadata.Full = false
		consCfg.Version = consVer
		c, err := NewConsumer(FunctionalTestEnv.KafkaBrokerAddrs, consCfg)
		if err != nil {
			t.Fatal(err)
		}
		defer safeClose(t, c) //nolint: gocritic // the close intentionally happens outside the loop
		pc, err := c.ConsumePartition("test.1", 0, producedMessages[0].Offset)
		if err != nil {
			t.Fatal(err)
		}
		defer safeClose(t, pc) //nolint: gocritic // the close intentionally happens outside the loop

		var wg sync.WaitGroup
		wg.Add(1)
		g.Go(func() error {
			// Consume as many messages as there have been produced and make sure that
			// order is preserved.
			t.Logf("*** Consuming with client version %s\n", consVer)
			for i, prodMsg := range producedMessages {
				select {
				case consMsg := <-pc.Messages():
					if consMsg.Offset != prodMsg.Offset {
						t.Fatalf("Consumed unexpected offset: version=%s, index=%d, want=%s, got=%s",
							consVer, i, prodMsg2Str(prodMsg), consMsg2Str(consMsg))
					}
					if string(consMsg.Value) != string(prodMsg.Value.(StringEncoder)) {
						t.Fatalf("Consumed unexpected msg: version=%s, index=%d, want=%s, got=%s",
							consVer, i, prodMsg2Str(prodMsg), consMsg2Str(consMsg))
					}
					if i == 0 {
						t.Logf("Consumed first msg: version=%s, index=%d, got=%s",
							consVer, i, consMsg2Str(consMsg))
						wg.Done()
					}
					if i%1000 == 0 {
						t.Logf("Consumed messages: version=%s, index=%d, got=%s",
							consVer, i, consMsg2Str(consMsg))
					}
				case <-time.After(15 * time.Second):
					t.Fatalf("Timeout %s waiting for: index=%d, offset=%d, msg=%s",
						consCfg.ClientID, i, prodMsg.Offset, prodMsg.Value)
				}
			}
			return nil
		})
		wg.Wait() // wait for first message to be consumed before starting next consumer
	}