Function readOffset

writer_test.go:407–424

readOffset gets the latest offset for the given topic/partition.

(topic string, partition int)

// readOffset gets the latest offset for the given topic/partition.
func readOffset(topic string, partition int) (offset int64, err error) {
	var conn *Conn

	ctx, cancel := context.WithTimeout(context.Background(), 10*time.Second)
	defer cancel()

	if conn, err = DialLeader(ctx, "tcp", "localhost:9092", topic, partition); err != nil {
		err = fmt.Errorf("readOffset, DialLeader: %w", err)
		return
	}
	defer conn.Close()

	offset, err = conn.ReadLastOffset()
	if err != nil {
		err = fmt.Errorf("readOffset, conn.ReadLastOffset: %w", err)
	}
	return
}

func readPartition(topic string, partition int, offset int64) (msgs []Message, err error) {
	var conn *Conn