func (s *ServiceHTTPSuite) TestSetOffsetsInvalidPartition(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Post("http://_/topics/test.4/offsets?group=foo",
		"application/json", strings.NewReader(`[{"partition": 5}]`))
	c.Check(err, IsNil)
	body := ParseJSONBody(c, r).(map[string]interface{})
	if s.proxyCfg.Kafka.Version.IsAtLeast(sarama.V2_0_0_0) {
		c.Check(r.StatusCode, Equals, http.StatusNotFound)
		c.Check(body["error"], Equals, "Unknown topic")
	} else {
		c.Check(r.StatusCode, Equals, http.StatusOK)
		c.Check(body, DeepEquals, httpsrv.EmptyResponse)
	}
}
It is not an error to set an offset for a missing partition.
package service
import (
	"bytes"
	"encoding/base64"
	"encoding/json"
	"fmt"
	"io"
	"io/ioutil"
	"net"
	"net/http"
	"os"
	"path"
	"sort"
	"strconv"
	"strings"
	"sync"
	"time"
	"github.com/Shopify/sarama"
	"github.com/mailgun/kafka-pixy/actor"
	"github.com/mailgun/kafka-pixy/config"
	pb "github.com/mailgun/kafka-pixy/gen/golang"
	"github.com/mailgun/kafka-pixy/server/httpsrv"
	"github.com/mailgun/kafka-pixy/testhelpers"
	"github.com/mailgun/kafka-pixy/testhelpers/kafkahelper"
	"github.com/pkg/errors"
	. "gopkg.in/check.v1"
)
type ServiceHTTPSuite struct {
	ns         *actor.Descriptor
	cfg        *config.App
	proxyCfg   *config.Proxy
	kh         *kafkahelper.T
	unixClient *http.Client
	tcpClient  *http.Client
}
var _ = Suite(&ServiceHTTPSuite{})
func (s *ServiceHTTPSuite) SetUpSuite(c *C) {
	testhelpers.InitLogging()
}
func (s *ServiceHTTPSuite) SetUpTest(c *C) {
	s.ns = actor.Root().NewChild("T")
	s.cfg = &config.App{Proxies: make(map[string]*config.Proxy)}
	s.cfg.TCPAddr = "127.0.0.1:19092"
	s.cfg.UnixAddr = path.Join(os.TempDir(), "kafka-pixy.sock")
	s.proxyCfg = testhelpers.NewTestProxyCfg("pxyH_client_id")
	s.proxyCfg.Consumer.OffsetsCommitInterval = 50 * time.Millisecond
	s.cfg.Proxies["pxyH"] = s.proxyCfg
	s.cfg.DefaultCluster = "pxyH"
	os.Remove(s.cfg.UnixAddr)
	s.kh = kafkahelper.New(c)
	s.unixClient = testhelpers.NewUDSHTTPClient(s.cfg.UnixAddr)
	s.tcpClient = &http.Client{Transport: &http.Transport{
		DialContext: (&net.Dialer{
			Timeout:   30 * time.Second,
			KeepAlive: 30 * time.Second,
		}).DialContext,
	}}
}
func (s *ServiceHTTPSuite) TearDownTest(c *C) {
	s.kh.Close()
}
func (s *ServiceHTTPSuite) TestInvalidUnixAddr(c *C) {
	s.cfg.TCPAddr = "127.0.0.1:19099"
	s.cfg.UnixAddr = "/tmp"
	svc, err := Spawn(s.cfg)
	c.Check(err.Error(), Equals, "failed to start Unix socket based HTTP API server: "+
		"failed to create listener: "+
		"listen unix /tmp: "+
		"bind: address already in use")
	c.Check(svc, IsNil)
}
func (s *ServiceHTTPSuite) TestInvalidKafkaPeers(c *C) {
	s.cfg.Proxies[s.cfg.DefaultCluster].Kafka.SeedPeers = []string{"localhost:12345"}
	svc, err := Spawn(s.cfg)
	c.Check(err.Error(), Equals, "failed to spawn proxy, name=pxyH: "+
		"failed to create Kafka client: "+
		"kafka: client has run out of available brokers to talk to (Is your cluster reachable?)")
	c.Check(svc, IsNil)
}
func (s *ServiceHTTPSuite) TestProduce(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	for i := 0; i < 10; i++ {
		s.unixClient.Post("http://_/topics/test.4/messages?key=1",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
		s.unixClient.Post("http://_/topics/test.4/messages?key=2",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
		s.unixClient.Post("http://_/topics/test.4/messages?key=3",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
		s.unixClient.Post("http://_/topics/test.4/messages?key=4",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
		s.unixClient.Post("http://_/topics/test.4/messages?key=5",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
	}
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	c.Check(offsetsAfter[0], Equals, offsetsBefore[0]+20)
	c.Check(offsetsAfter[1], Equals, offsetsBefore[1]+10)
	c.Check(offsetsAfter[2], Equals, offsetsBefore[2]+10)
	c.Check(offsetsAfter[3], Equals, offsetsBefore[3]+10)
}
func (s *ServiceHTTPSuite) TestProduceNilKey(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	for i := 0; i < 100; i++ {
		s.unixClient.Post("http://_/topics/test.4/messages",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
	}
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	delta0 := offsetsAfter[0] - offsetsBefore[0]
	delta1 := offsetsAfter[1] - offsetsBefore[1]
	if delta0 == 0 || delta1 == 0 {
		panic(errors.Errorf("Too high imbalance: %v != %v", delta0, delta1))
	}
}
func (s *ServiceHTTPSuite) TestProduceEmptyKey(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	for i := 0; i < 10; i++ {
		s.unixClient.Post("http://_/topics/test.4/messages?key=",
			"text/plain", strings.NewReader(strconv.Itoa(i)))
	}
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	c.Check(offsetsAfter[0], Equals, offsetsBefore[0])
	c.Check(offsetsAfter[1], Equals, offsetsBefore[1])
	c.Check(offsetsAfter[2], Equals, offsetsBefore[2])
	c.Check(offsetsAfter[3], Equals, offsetsBefore[3]+10)
}
func (s *ServiceHTTPSuite) TestUtf8Message(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	s.unixClient.Post("http://_/topics/test.4/messages?key=foo",
		"text/plain", strings.NewReader("Превед Медвед"))
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	msgs := s.kh.GetMessages("test.4", offsetsBefore, offsetsAfter)
	c.Check(msgs, DeepEquals,
		[][]string{[]string(nil), {"Превед Медвед"}, []string(nil), []string(nil)})
}
func (s *ServiceHTTPSuite) TestProduceHeaders(c *C) {
	if !s.proxyCfg.Kafka.Version.IsAtLeast(sarama.V0_11_0_0) {
		c.Skip("Headers not supported before Kafka v0.11")
	}
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.1")
	req, err := http.NewRequest("POST", "http://_/topics/test.1/messages?key=foo&sync",
		strings.NewReader("test"))
	c.Assert(err, IsNil)
	req.Header.Add("Content-Type", "text/plain")
	req.Header.Add("X-Kafka-Foo", base64.StdEncoding.EncodeToString([]byte("bar")))
	rs, err := s.unixClient.Do(req)
	c.Check(err, IsNil)
	c.Check(rs.StatusCode, Equals, http.StatusOK)
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.1")
	c.Check(offsetsAfter[0], Equals, offsetsBefore[0]+1)
}
func (s *ServiceHTTPSuite) TestProduceInvalidHeaders(c *C) {
	if !s.proxyCfg.Kafka.Version.IsAtLeast(sarama.V0_11_0_0) {
		c.Skip("Headers not supported before Kafka v0.11")
	}
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	req, err := http.NewRequest("POST", "http://_/topics/test.1/messages?key=foo&sync",
		strings.NewReader("test"))
	c.Assert(err, IsNil)
	req.Header.Add("Content-Type", "text/plain")
	req.Header.Add("X-Kafka-Foo", "invalid!")
	rs, err := s.unixClient.Do(req)
	c.Check(err, IsNil)
	c.Check(rs.StatusCode, Equals, http.StatusBadRequest)
	body := ParseJSONBody(c, rs).(map[string]interface{})
	c.Check(body["error"], Equals, "Invalid base64 encoding for header: X-Kafka-Foo")
}
func (s *ServiceHTTPSuite) TestProduceHeadersUnsupported(c *C) {
	if s.proxyCfg.Kafka.Version.IsAtLeast(sarama.V0_11_0_0) {
		c.Skip("Headers are supported after Kafka v0.11")
	}
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	req, err := http.NewRequest("POST", "http://_/topics/test.1/messages?key=foo&sync",
		strings.NewReader("test"))
	c.Assert(err, IsNil)
	req.Header.Add("Content-Type", "text/plain")
	req.Header.Add("X-Kafka-Foo", base64.StdEncoding.EncodeToString([]byte("bar")))
	rs, err := s.unixClient.Do(req)
	c.Check(err, IsNil)
	c.Check(rs.StatusCode, Equals, http.StatusBadRequest)
	body := ParseJSONBody(c, rs).(map[string]interface{})
	c.Check(body["error"], Matches, "headers are not supported with this version of Kafka.*")
}
func (s *ServiceHTTPSuite) TestProduceXWWWFormUrlencoded(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.1")
	rs, err := s.unixClient.Post("http://_/topics/test.1/messages?key=1",
		"application/x-www-form-urlencoded", strings.NewReader("msg=foo"))
	c.Check(err, IsNil)
	c.Check(rs.StatusCode, Equals, http.StatusOK)
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.1")
	c.Check(offsetsAfter[0], Equals, offsetsBefore[0]+1)
}
func (s *ServiceHTTPSuite) TestBothAPI(c *C) {
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	s.cfg.TCPAddr = "127.0.0.1:55501"
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	_, err1 := s.tcpClient.Post("http://localhost:55501/topics/test.4/messages?key=foo",
		"text/plain", strings.NewReader("Превед"))
	_, err2 := s.unixClient.Post("http://_/topics/test.4/messages?key=foo",
		"text/plain", strings.NewReader("Kitty"))
	svc.Stop()  
	c.Check(err1, IsNil)
	c.Check(err2, IsNil)
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	fmt.Printf("*** Before=%v, After=%v\n", offsetsBefore, offsetsAfter)
	msgs := s.kh.GetMessages("test.4", offsetsBefore, offsetsAfter)
	c.Check(msgs, DeepEquals,
		[][]string{[]string(nil), {"Превед", "Kitty"}, []string(nil), []string(nil)})
}
func (s *ServiceHTTPSuite) TestStoppedServerCall(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	_, err = s.unixClient.Post("http://_/topics/test.4/messages?key=foo",
		"text/plain", strings.NewReader("Hello"))
	c.Check(err, IsNil)
	svc.Stop()
	r, err := s.unixClient.Post("http://_/topics/test.4/messages?key=foo",
		"text/plain", strings.NewReader("Kitty"))
	c.Check(err.Error(), Matches, "Post http://_/topics/test\\.4/messages\\?key=foo: dial unix .* no such file or directory")
	c.Check(r, IsNil)
}
func (s *ServiceHTTPSuite) TestLargestMessage(c *C) {
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	saramaCfg := s.proxyCfg.SaramaProducerCfg()
	maxMsgSize := saramaCfg.Producer.MaxMessageBytes
	if saramaCfg.Version.IsAtLeast(sarama.V0_11_0_0) {
		maxMsgSize -= 39
	} else {
		maxMsgSize -= ProdMsgMetadataSize([]byte("foo"))
	}
	msg := GenMessage(maxMsgSize)
	s.cfg.TCPAddr = "127.0.0.1:55501"
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	r := PostChunked(s.tcpClient, "http://127.0.0.1:55501/topics/test.4/messages?key=foo", msg)
	svc.Stop()  
	c.Check(r.StatusCode, Equals, http.StatusOK)
	c.Check(ParseJSONBody(c, r), DeepEquals, map[string]interface{}{})
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	messages := s.kh.GetMessages("test.4", offsetsBefore, offsetsAfter)
	readMsg := messages[1][0]
	c.Check(readMsg, Equals, msg)
}
func (s *ServiceHTTPSuite) TestMessageTooLarge(c *C) {
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	maxMsgSize := sarama.NewConfig().Producer.MaxMessageBytes - ProdMsgMetadataSize([]byte("foo")) + 1
	msg := GenMessage(maxMsgSize)
	s.cfg.TCPAddr = "127.0.0.1:55501"
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	r := PostChunked(s.tcpClient, "http://127.0.0.1:55501/topics/test.4/messages?key=foo", msg)
	svc.Stop()  
	c.Check(r.StatusCode, Equals, http.StatusOK)
	c.Check(ParseJSONBody(c, r), DeepEquals, map[string]interface{}{})
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	c.Check(offsetsAfter, DeepEquals, offsetsBefore)
}
func (s *ServiceHTTPSuite) TestSyncProduce(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	offsetsBefore := s.kh.GetNewestOffsets("test.4")
	r, err := s.unixClient.Post("http://_/topics/test.4/messages?key=1&sync",
		"text/plain", strings.NewReader("Foo"))
	svc.Stop()  
	offsetsAfter := s.kh.GetNewestOffsets("test.4")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusOK)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(int(body["partition"].(float64)), Equals, 0)
	c.Check(int64(body["offset"].(float64)), Equals, offsetsBefore[0])
	c.Check(offsetsAfter[0], Equals, offsetsBefore[0]+1)
}
func (s *ServiceHTTPSuite) TestSyncProduceInvalidTopic(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Post("http://_/topics/no-such-topic/messages?sync=true",
		"text/plain", strings.NewReader("Foo"))
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusNotFound)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, sarama.ErrUnknownTopicOrPartition.Error())
}
func (s *ServiceHTTPSuite) TestConsumeNoGroup(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Get("http://_/topics/test.4/messages")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusBadRequest)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "one consumer group is expected, but 0 provided")
}
func (s *ServiceHTTPSuite) TestConsumeManyGroups(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Get("http://_/topics/test.4/messages?group=a&group=b")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusBadRequest)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "one consumer group is expected, but 2 provided")
}
func (s *ServiceHTTPSuite) TestConsumeInvalidTopic(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Get("http://_/topics/no-such-topic/messages?group=foo")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusRequestTimeout)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "long polling timeout")
}
func (s *ServiceHTTPSuite) TestConsumeAutoAck(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	s.kh.ResetOffsets("foo", "test.4")
	produced := s.kh.PutMessages("auto-ack", "test.4", map[string]int{"A": 17, "B": 19, "C": 23, "D": 29})
	consumed := make(map[string][]*pb.ConsRs)
	offsetsBefore := s.kh.GetCommittedOffsets("foo", "test.4")
	for i := 0; i < 88; i++ {
		res, err := s.unixClient.Get("http://_/topics/test.4/messages?group=foo")
		c.Check(err, IsNil, Commentf("failed to consume message #%d", i))
		consRes := ParseConsRes(c, res)
		key := string(consRes.KeyValue)
		consumed[key] = append(consumed[key], consRes)
	}
	svc.Stop()
	offsetsAfter := s.kh.GetCommittedOffsets("foo", "test.4")
	c.Check(offsetsAfter[0].Val, Equals, offsetsBefore[0].Val+17)
	c.Check(offsetsAfter[1].Val, Equals, offsetsBefore[1].Val+29)
	c.Check(offsetsAfter[2].Val, Equals, offsetsBefore[2].Val+23)
	c.Check(offsetsAfter[3].Val, Equals, offsetsBefore[3].Val+19)
	assertMsgs(c, consumed, produced)
}
func (s *ServiceHTTPSuite) TestConsumeNoAck(c *C) {
	s.proxyCfg.Consumer.AckTimeout = 500 * time.Millisecond
	s.proxyCfg.Consumer.SubscriptionTimeout = 500 * time.Millisecond
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	s.kh.ResetOffsets("foo", "test.1")
	s.kh.PutMessages("no-ack", "test.1", map[string]int{"A": 1})
	offsetsBefore := s.kh.GetCommittedOffsets("foo", "test.1")
	_, err = s.unixClient.Get("http://_/topics/test.1/messages?group=foo&noAck")
	c.Check(err, IsNil)
	svc.Stop()
	offsetsAfter := s.kh.GetCommittedOffsets("foo", "test.1")
	c.Check(offsetsAfter[0].Val, Equals, offsetsBefore[0].Val)
}
func (s *ServiceHTTPSuite) TestConsumeExplicitAck(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	s.kh.ResetOffsets("foo", "test.4")
	produced := s.kh.PutMessages("explicit-ack", "test.4", map[string]int{"A": 17, "B": 19, "C": 23, "D": 29})
	consumed := make(map[string][]*pb.ConsRs)
	offsetsBefore := s.kh.GetCommittedOffsets("foo", "test.4")
	res, err := s.unixClient.Get("http://_/topics/test.4/messages?group=foo&noAck")
	c.Check(err, IsNil, Commentf("failed to consume first message"))
	consRes := ParseConsRes(c, res)
	key := string(consRes.KeyValue)
	consumed[key] = append(consumed[key], consRes)
	for i := 1; i < 88; i++ {
		url := fmt.Sprintf("http://_/topics/test.4/messages?group=foo&ackPartition=%d&ackOffset=%d",
			consRes.Partition, consRes.Offset)
		httpRs, err := s.unixClient.Get(url)
		consRes = ParseConsRes(c, httpRs)
		c.Check(err, IsNil, Commentf("failed to consume message #%d", i))
		c.Check(httpRs.StatusCode, Equals, http.StatusOK)
		key := string(consRes.KeyValue)
		consumed[key] = append(consumed[key], consRes)
	}
	url := fmt.Sprintf("http://_/topics/test.4/acks?group=foo&partition=%d&offset=%d",
		consRes.Partition, consRes.Offset)
	httpRs, err := s.unixClient.Post(url, "text/plain", nil)
	c.Check(err, IsNil, Commentf("failed ack last message"))
	ackRs := ParseJSONBody(c, httpRs).(map[string]interface{})
	c.Check(ackRs, DeepEquals, map[string]interface{}{})
	c.Check(httpRs.StatusCode, Equals, http.StatusOK)
	svc.Stop()
	offsetsAfter := s.kh.GetCommittedOffsets("foo", "test.4")
	c.Check(offsetsAfter[0].Val, Equals, offsetsBefore[0].Val+17)
	c.Check(offsetsAfter[1].Val, Equals, offsetsBefore[1].Val+29)
	c.Check(offsetsAfter[2].Val, Equals, offsetsBefore[2].Val+23)
	c.Check(offsetsAfter[3].Val, Equals, offsetsBefore[3].Val+19)
	assertMsgs(c, consumed, produced)
}
func (s *ServiceHTTPSuite) TestConsumeDisabled(c *C) {
	s.proxyCfg.Consumer.Disabled = true
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Get("http://_/topics/test.4/messages?group=foo")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusServiceUnavailable)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "service is disabled by configuration")
}
func (s *ServiceHTTPSuite) TestConsumeHeaders(c *C) {
	if !s.proxyCfg.Kafka.Version.IsAtLeast(sarama.V0_11_0_0) {
		c.Skip("Headers not supported before Kafka v0.11")
	}
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	s.kh.ResetOffsets("foo", "test.4")
	req, err := http.NewRequest("POST", "http://_/topics/test.4/messages?key=foo&sync",
		strings.NewReader("test"))
	c.Assert(err, IsNil)
	req.Header.Add("Content-Type", "text/plain")
	req.Header.Add("X-Kafka-Foo", base64.StdEncoding.EncodeToString([]byte("bar")))
	rs, err := s.unixClient.Do(req)
	c.Assert(err, IsNil)
	c.Assert(rs.StatusCode, Equals, http.StatusOK)
	res, err := s.unixClient.Get("http://_/topics/test.4/messages?group=foo")
	c.Check(err, IsNil)
	consRes := ParseConsRes(c, res)
	svc.Stop()
	c.Check(consRes.Headers, Not(HasLen), 0)
	c.Check(consRes.Headers[0], DeepEquals, &pb.RecordHeader{Key: "Foo", Value: []byte("bar")})
}
func (s *ServiceHTTPSuite) TestGetOffsetsNoSuchGroup(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Get("http://_/topics/test.4/offsets?group=no_such_group")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusOK)
	body := ParseJSONBody(c, r).([]interface{})
	for i := 0; i < 4; i++ {
		partitionView := body[i].(map[string]interface{})
		c.Check(partitionView["partition"].(float64), Equals, float64(i))
		c.Check(partitionView["offset"].(float64), Equals, float64(-1))
	}
}
func (s *ServiceHTTPSuite) TestGetOffsetsNoSuchTopic(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Get("http://_/topics/no_such_topic/offsets?group=foo")
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusNotFound)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "Unknown topic")
}
func (s *ServiceHTTPSuite) TestSetOffsets(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Post("http://_/topics/test.4/offsets?group=foo",
		"application/json", strings.NewReader(
			`[{"partition": 0, "offset": 1100, "metadata": "A100"},
			  {"partition": 1, "offset": 1101, "metadata": "A101"},
			  {"partition": 2, "offset": 1102, "metadata": "A102"},
			  {"partition": 3, "offset": 1103, "metadata": "A103"}]`))
	c.Check(err, IsNil)
	c.Check(ParseJSONBody(c, r), DeepEquals, httpsrv.EmptyResponse)
	c.Check(r.StatusCode, Equals, http.StatusOK)
	r, err = s.unixClient.Get("http://_/topics/test.4/offsets?group=foo")
	c.Check(err, IsNil)
	body := ParseJSONBody(c, r).([]interface{})
	for i := 0; i < 4; i++ {
		partitionView := body[i].(map[string]interface{})
		c.Check(partitionView["partition"].(float64), Equals, float64(i))
		c.Check(partitionView["offset"].(float64), Equals, float64(1100+i))
		c.Check(partitionView["metadata"].(string), Equals, fmt.Sprintf("A10%d", i))
		c.Check(partitionView["count"], Equals, partitionView["end"].(float64)-partitionView["begin"].(float64))
	}
}
func (s *ServiceHTTPSuite) TestSetOffsetsNoSuchTopic(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Post("http://_/topics/no_such_topic/offsets?group=foo",
		"application/json", strings.NewReader(`[{"partition": 0, "offset": 1100, "metadata": "A100"}]`))
	kafkaVersion := os.Getenv("KAFKA_VERSION")
	if strings.HasPrefix(kafkaVersion, "0.8") {
		c.Check(err, IsNil)
		c.Check(r.StatusCode, Equals, http.StatusOK)
		c.Check(ParseJSONBody(c, r), DeepEquals, httpsrv.EmptyResponse)
		return
	}
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusNotFound)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "Unknown topic")
}
func (s *ServiceHTTPSuite) TestSetOffsetsInvalidBody(c *C) {
	svc, err := Spawn(s.cfg)
	c.Assert(err, IsNil)
	defer svc.Stop()
	r, err := s.unixClient.Post("http://_/topics/test.4/offsets?group=foo",
		"application/json", strings.NewReader(`garbage`))
	c.Check(err, IsNil)
	c.Check(r.StatusCode, Equals, http.StatusBadRequest)
	body := ParseJSONBody(c, r).(map[string]interface{})
	c.Check(body["error"], Equals, "Failed to parse the request: err=(invalid character 'g' looking for beginning of value)")
}