单元测试 Sarama Kafka 库中消费者组的 session.MarkMessage() 方法

在IT行业这个发展更新速度很快的行业，只有不停止的学习，才不会被行业所淘汰。如果你是Golang学习者，那么本文《单元测试 Sarama Kafka 库中消费者组的 session.MarkMessage() 方法》就很适合你！本篇内容主要包括##content_title##，希望对大家的知识积累有所帮助，助力实战开发！

我正在尝试调整 github.com/shopify/sarama 的消费者组示例中的代码，并且正在努力添加一个单元测试来测试 session.markmessage() 中 consumeclaim 方法的功能（https： //github.com/shopify/sarama/blob/5466b37850a38f4ed6d04b94c6f058bd75032c2a/examples/consumergroup/main.go#l160)。

这是我使用 consume() 函数改编的代码：

package main

import (
    "context"
    "fmt"
    "log"
    "os"
    "os/signal"
    "sync"
    "syscall"

    "github.com/shopify/sarama"
)

var (
    addrs = []string{"localhost:9092"}
    topic = "my-topic"
)

func main() {
    ctx, cancel := context.withcancel(context.background())
    defer cancel()

    var wg sync.waitgroup
    defer wg.wait()

    consumer := &consumer{ready: make(chan bool)}

    close := consume(ctx, &wg, consumer)
    defer close()

    <-consumer.ready
    log.println("sarama consumer up and running!")

    sigterm := make(chan os.signal, 1)
    signal.notify(sigterm, syscall.sigint, syscall.sigterm)
    select {
    case <-ctx.done():
        log.println("terminating: context cancelled")
    case <-sigterm:
        log.println("terminating: via signal")
    }
}

func consume(ctx context.context, wg *sync.waitgroup, consumer *consumer) (close func()) {
    config := sarama.newconfig()
    config.version = sarama.v0_11_0_2 // the version has to be at least v0_10_2_0 to support consumer groups
    config.consumer.offsets.initial = sarama.offsetoldest

    consumergroup, err := sarama.newconsumergroup(addrs, "my-group", config)
    if err != nil {
        log.fatalf("newconsumergroup: %v", err)
    }

    wg.add(1)
    go func() {
        defer wg.done()
        for {
            if err := consumergroup.consume(ctx, []string{topic}, consumer); err != nil {
                log.panicf("consume: %v", err)
            }
            if ctx.err() != nil {
                return
            }
            consumer.ready = make(chan bool)
        }
    }()

    close = func() {
        if err := consumergroup.close(); err != nil {
            log.panicf("close: %v", err)
        }
    }
    return
}

// consumer represents a sarama consumer group consumer
type consumer struct {
    ready  chan bool
    handle func([]byte) error
}

// setup is run at the beginning of a new session, before consumeclaim
func (consumer *consumer) setup(sarama.consumergroupsession) error {
    // mark the consumer as ready
    close(consumer.ready)
    return nil
}

// cleanup is run at the end of a session, once all consumeclaim goroutines have exited
func (consumer *consumer) cleanup(sarama.consumergroupsession) error {
    return nil
}

// consumeclaim must start a consumer loop of consumergroupclaim's messages().
func (consumer *consumer) consumeclaim(session sarama.consumergroupsession, claim sarama.consumergroupclaim) error {
    for message := range claim.messages() {
        log.printf("message claimed: value = %s, timestamp = %v, topic = %s", message.value, message.timestamp, message.topic)
        if consumer.handle != nil {
            if err := consumer.handle(message.value); err != nil {
                return fmt.errorf("handle message %s: %v", message.value, err)
            }
        }
        session.markmessage(message, "")
    }
    return nil
}

以下是我为其编写的几个单元测试：

package main

import (
    "context"
    "fmt"
    "log"
    "sync"
    "testing"
    "time"

    "github.com/shopify/sarama"
    "github.com/stretchr/testify/require"
    "gotest.tools/assert"
)

func testconsume(t *testing.t) {
    config := sarama.newconfig()
    config.producer.return.successes = true
    producer, err := sarama.newsyncproducer(addrs, config)
    require.noerror(t, err)

    partition, offset, err := producer.sendmessage(&sarama.producermessage{
        topic: topic,
        value: sarama.byteencoder([]byte("foobar")),
    })
    require.noerror(t, err)
    t.logf("sent message to partition %d with offset %d", partition, offset)

    ctx, cancel := context.withcancel(context.background())
    var wg sync.waitgroup

    consumer := &consumer{ready: make(chan bool)}

    close := consume(ctx, &wg, consumer)

    <-consumer.ready
    log.println("sarama consumer up and running!")

    time.sleep(1 * time.second)

    cancel()
    wg.wait()
    close()
}

func testconsumetwice(t *testing.t) {
    config := sarama.newconfig()
    config.producer.return.successes = true
    producer, err := sarama.newsyncproducer(addrs, config)
    require.noerror(t, err)

    data1, data2 := "foobar1", "foobar2"

    for _, data := range []string{data1, data2} {
        partition, offset, err := producer.sendmessage(&sarama.producermessage{
            topic: topic,
            key:   sarama.stringencoder("foobar"),
            value: sarama.stringencoder(data),
        })
        require.noerror(t, err)
        t.logf("sent message to partition %d with offset %d", partition, offset)
    }

    ctx, cancel := context.withcancel(context.background())
    var wg sync.waitgroup

    messagereceived := make(chan []byte)
    consumer := &consumer{
        ready: make(chan bool),
        handle: func(data []byte) error {
            messagereceived <- data
            fmt.printf("received message: %s\n", data)
            return nil
        },
    }

    close := consume(ctx, &wg, consumer)

    <-consumer.ready
    log.println("sarama consumer up and running!")

    for i := 0; i < 2; i++ {
        data := <-messagereceived
        switch i {
        case 0:
            assert.equal(t, data1, string(data))
        case 1:
            assert.equal(t, data2, string(data))
        }
    }

    cancel()
    wg.wait()
    close()
}

可以在 docker 容器（例如 johnnypark/kafka-zookeeper）中运行 kafka 和 zookeeper 后运行测试，如下所示：

docker run -p 2181:2181 -p 9092:9092 -e advertised_host=127.0.0.1  -e num_partitions=10 johnnypark/kafka-zookeeper

我正在努力解决以下问题：如果我注释掉该行

        session.MarkMessage(message, "")

测试仍然通过。根据 https://godoc.org/github.com/shopify/sarama#consumergroupsession，markmessage 将消息标记为已使用，但我如何在单元测试中测试它？

解决方案

sarama.ConsumerGroupSession.MarkMessage 调用 sarama.PartitionOffsetManager.MarkOffset，并在方法注释中说：“注意：出于效率原因，调用 MarkOffset 不一定会立即将偏移量提交到后端存储，并且可能永远不会提交，如果您的应用程序崩溃了。这意味着您最终可能会处理同一消息两次。”

因此，在单元测试中，MarkMessage 提交偏移量的速度不够快。我遇到了同样的问题，谷歌把我带到了这里。在测试函数结束时休眠一秒钟可能是一种解决方法。

以上就是《单元测试 Sarama Kafka 库中消费者组的 session.MarkMessage() 方法》的详细内容，更多关于的资料请关注golang学习网公众号！