refactor & add offset retention.

h12w · Sep 17, 2015 · 58f6fac · 58f6fac
1 parent 41a830d
commit 58f6fac
Show file tree

Hide file tree

Showing 8 changed files with 34 additions and 49 deletions.
diff --git a/TODO.md b/TODO.md
@@ -7,8 +7,12 @@ Design principle
 * Stateless
 * fail fast
   - timeout (SetDealine)
+  - release resources
 * fault tolerance:
-  - automatic handling of broken connections
-  - leader down
+  - try again for broken connection (maxBadConnRetries=2)
+  - recover without restart
+      - broken connections
+      - leader down
+  - retry
   - partition expand
   - graceful shutdown
diff --git a/broker/broker.go b/broker/broker.go
@@ -15,7 +15,6 @@ var (
 
 type Config struct {
 	Addr         string
-	SendQueueLen int
 	RecvQueueLen int
 	Timeout      time.Duration
 }
@@ -42,7 +41,7 @@ func New(config *Config) (*B, error) {
 	b := &B{
 		config:   config,
 		conn:     conn,
-		sendChan: make(chan *brokerJob, config.SendQueueLen),
+		sendChan: make(chan *brokerJob),
 		recvChan: make(chan *brokerJob, config.RecvQueueLen),
 	}
 	go b.sendLoop()
@@ -54,12 +53,12 @@ func (b *B) Close() {
 	b.conn.Close()
 }
 
-func (b *B) Do(req *proto.Request, resp *proto.Response) error {
+func (b *B) Do(req *proto.Request, resp proto.ResponseMessage) error {
 	req.CorrelationID = atomic.AddInt32(&b.cid, 1)
 	errChan := make(chan error)
 	b.sendChan <- &brokerJob{
 		req:     req,
-		resp:    resp,
+		resp:    &proto.Response{ResponseMessage: resp},
 		errChan: errChan,
 	}
 	return <-errChan

diff --git a/broker/broker_test.go b/broker/broker_test.go
@@ -16,7 +16,6 @@ const (
 func TestMeta(t *testing.T) {
 	broker, err := New(&Config{
 		Addr:         kafkaAddr,
-		SendQueueLen: 10,
 		RecvQueueLen: 10,
 		Timeout:      time.Second,
 	})
@@ -32,19 +31,16 @@ func TestMeta(t *testing.T) {
 			"test",
 		},
 	}
-	resp := &proto.Response{
-		ResponseMessage: &proto.TopicMetadataResponse{},
-	}
+	resp := &proto.TopicMetadataResponse{}
 	if err := broker.Do(req, resp); err != nil {
 		t.Fatal(err)
 	}
-	fmt.Println(toJSON(resp.ResponseMessage))
+	fmt.Println(toJSON(resp))
 }
 
 func TestConsumeAll(t *testing.T) {
 	broker, err := New(&Config{
 		Addr:         kafkaAddr,
-		SendQueueLen: 10,
 		RecvQueueLen: 10,
 		Timeout:      time.Second,
 	})
@@ -75,7 +71,7 @@ func TestConsumeAll(t *testing.T) {
 		},
 	}
 	resp := proto.FetchResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		t.Fatal(err)
 	}
 	fmt.Println(toJSON(resp))
@@ -91,7 +87,6 @@ func TestConsumeAll(t *testing.T) {
 func TestOffsetCommit(t *testing.T) {
 	broker, err := New(&Config{
 		Addr:         kafkaAddr,
-		SendQueueLen: 10,
 		RecvQueueLen: 10,
 		Timeout:      time.Second,
 	})
@@ -124,7 +119,7 @@ func TestOffsetCommit(t *testing.T) {
 		},
 	}
 	resp := proto.OffsetCommitResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		t.Fatal(t)
 	}
 	fmt.Println(toJSON(resp))
@@ -133,7 +128,6 @@ func TestOffsetCommit(t *testing.T) {
 func TestOffsetFetch(t *testing.T) {
 	broker, err := New(&Config{
 		Addr:         kafkaAddr,
-		SendQueueLen: 10,
 		RecvQueueLen: 10,
 		Timeout:      time.Second,
 	})
@@ -156,7 +150,7 @@ func TestOffsetFetch(t *testing.T) {
 		},
 	}
 	resp := proto.OffsetFetchResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		t.Fatal(err)
 	}
 	fmt.Println(toJSON(resp))
@@ -165,7 +159,6 @@ func TestOffsetFetch(t *testing.T) {
 func TestConsumerMeta(t *testing.T) {
 	broker, err := New(&Config{
 		Addr:         kafkaAddr,
-		SendQueueLen: 10,
 		RecvQueueLen: 10,
 		Timeout:      time.Second,
 	})
@@ -181,7 +174,7 @@ func TestConsumerMeta(t *testing.T) {
 		RequestMessage: &creq,
 	}
 	resp := proto.ConsumerMetadataResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		t.Fatal(err)
 	}
 	fmt.Println(toJSON(resp))
@@ -190,7 +183,6 @@ func TestConsumerMeta(t *testing.T) {
 func TestProduce(t *testing.T) {
 	broker, err := New(&Config{
 		Addr:         kafkaAddr,
-		SendQueueLen: 10,
 		RecvQueueLen: 10,
 		Timeout:      time.Second,
 	})
@@ -228,7 +220,7 @@ func TestProduce(t *testing.T) {
 		},
 	}
 	resp := proto.ProduceResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		t.Fatal(err)
 	}
 	fmt.Println(toJSON(resp))

diff --git a/client/client.go b/client/client.go
@@ -170,14 +170,9 @@ func (c *C) getTopicMetadata(topic string) (*proto.TopicMetadataResponse, error)
 	if needClosing {
 		defer broker.Close()
 	}
-	req := &proto.Request{
-		APIKey:         proto.TopicMetadataRequestType,
-		APIVersion:     0,
-		ClientID:       c.config.ClientID,
-		RequestMessage: &proto.TopicMetadataRequest{topic},
-	}
+	req := c.NewRequest(&proto.TopicMetadataRequest{topic})
 	resp := &proto.TopicMetadataResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: resp}); err != nil {
+	if err := broker.Do(req, resp); err != nil {
 		return nil, err
 	}
 	return resp, nil
@@ -192,14 +187,9 @@ func (c *C) getConsumerMetadata(consumerGroup string) (*proto.ConsumerMetadataRe
 		defer broker.Close()
 	}
 	creq := proto.ConsumerMetadataRequest(consumerGroup)
-	req := &proto.Request{
-		APIKey:         proto.ConsumerMetadataRequestType,
-		APIVersion:     0,
-		ClientID:       c.config.ClientID,
-		RequestMessage: &creq,
-	}
+	req := c.NewRequest(&creq)
 	resp := proto.ConsumerMetadataResponse{}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		return nil, err
 	}
 	return &resp, nil

diff --git a/consumer/consumer.go b/consumer/consumer.go
@@ -15,10 +15,11 @@ var (
 )
 
 type Config struct {
-	Client      client.Config
-	MaxWaitTime int32
-	MinBytes    int32
-	MaxBytes    int32
+	Client          client.Config
+	MaxWaitTime     int32
+	MinBytes        int32
+	MaxBytes        int32
+	OffsetRetention time.Duration
 }
 
 type C struct {
@@ -52,7 +53,7 @@ func (c *C) Offset(topic string, partition int32, consumerGroup string) (int64,
 	if err != nil {
 		return 0, err
 	}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		return 0, err
 	}
 	for i := range resp {
@@ -93,7 +94,7 @@ func (c *C) Consume(topic string, partition int32, offset int64) (values [][]byt
 	if err != nil {
 		return nil, err
 	}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		return nil, err
 	}
 	for i := range resp {
@@ -133,7 +134,7 @@ func (c *C) Commit(topic string, partition int32, consumerGroup string, offset i
 					{
 						Partition: partition,
 						Offset:    offset,
-						TimeStamp: time.Now().Unix(),
+						TimeStamp: time.Now().Add(c.config.OffsetRetention).Unix(),
 					},
 				},
 			},
@@ -144,7 +145,7 @@ func (c *C) Commit(topic string, partition int32, consumerGroup string, offset i
 	if err != nil {
 		return err
 	}
-	if err := broker.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := broker.Do(req, &resp); err != nil {
 		return err
 	}
 	for i := range resp {

diff --git a/consumer/consumer_test.go b/consumer/consumer_test.go
@@ -19,7 +19,7 @@ func TestGetOffset(t *testing.T) {
 	if err != nil {
 		t.Fatal(err)
 	}
-	fmt.Println("offset: ", offset)
+	fmt.Println("get offset: ", offset)
 }
 
 func TestConsumeAll(t *testing.T) {
@@ -50,14 +50,14 @@ func getConsumer(t *testing.T) *C {
 				"docker:32793",
 			},
 			BrokerConfig: broker.Config{
-				SendQueueLen: 10,
 				RecvQueueLen: 10,
 				Timeout:      time.Second,
 			},
 			ClientID: "abc",
 		},
-		MinBytes: 0,
-		MaxBytes: 9999,
+		MinBytes:        0,
+		MaxBytes:        9999,
+		OffsetRetention: 7 * 24 * time.Hour,
 	})
 	if err != nil {
 		t.Fatal(err)

diff --git a/producer/producer.go b/producer/producer.go
@@ -76,7 +76,7 @@ func (p *P) Produce(topic string, key, value []byte) error {
 		},
 	})
 	resp := proto.ProduceResponse{}
-	if err := leader.Do(req, &proto.Response{ResponseMessage: &resp}); err != nil {
+	if err := leader.Do(req, &resp); err != nil {
 		return err
 	}
 	for i := range resp {

diff --git a/producer/producer_test.go b/producer/producer_test.go
@@ -17,7 +17,6 @@ func TestProducer(t *testing.T) {
 				"docker:32793",
 			},
 			BrokerConfig: broker.Config{
-				SendQueueLen: 10,
 				RecvQueueLen: 10,
 				Timeout:      time.Second,
 			},