-
Notifications
You must be signed in to change notification settings - Fork 117
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #61 from mailgun/maxim/develop
Massive refactoring continues
- Loading branch information
Showing
26 changed files
with
1,464 additions
and
1,323 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| package consumer | ||
|
|
||
| type T interface { | ||
| // Consume consumes a message from the specified topic on behalf of the | ||
| // specified consumer group. If there are no more new messages in the topic | ||
| // at the time of the request then it will block for | ||
| // `Config.Consumer.LongPollingTimeout`. If no new message is produced during | ||
| // that time, then `ErrRequestTimeout` is returned. | ||
| // | ||
| // Note that during state transitions topic subscribe<->unsubscribe and | ||
| // consumer group register<->deregister the method may return either | ||
| // `ErrBufferOverflow` or `ErrRequestTimeout` even when there are messages | ||
| // available for consumption. In that case the user should back off a bit | ||
| // and then repeat the request. | ||
| Consume(group, topic string) (*Message, error) | ||
|
|
||
| // Stop sends a shutdown signal to all internal goroutines and blocks until | ||
| // they are stopped. It is guaranteed that all last consumed offsets of all | ||
| // consumer groups/topics are committed to Kafka before Consumer stops. | ||
| Stop() | ||
| } | ||
|
|
||
| // Message encapsulates a Kafka message returned by the consumer. | ||
| type Message struct { | ||
| Key, Value []byte | ||
| Topic string | ||
| Partition int32 | ||
| Offset int64 | ||
| HighWaterMark int64 | ||
| } | ||
|
|
||
| type ( | ||
| ErrSetup error | ||
| ErrBufferOverflow error | ||
| ErrRequestTimeout error | ||
| ) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,108 @@ | ||
| package consumerimpl | ||
|
|
||
| import ( | ||
| "fmt" | ||
| "time" | ||
|
|
||
| "github.com/Shopify/sarama" | ||
| "github.com/mailgun/kafka-pixy/actor" | ||
| "github.com/mailgun/kafka-pixy/config" | ||
| "github.com/mailgun/kafka-pixy/consumer" | ||
| "github.com/mailgun/kafka-pixy/consumer/dispatcher" | ||
| "github.com/mailgun/kafka-pixy/consumer/groupcsm" | ||
| "github.com/mailgun/kafka-pixy/consumer/offsetmgr" | ||
| "github.com/wvanbergen/kazoo-go" | ||
| ) | ||
|
|
||
| // T is a Kafka consumer implementation that automatically maintains consumer | ||
| // groups registrations and topic subscriptions. Whenever a message from a | ||
| // particular topic is consumed by a particular consumer group T checks if it | ||
| // has registered with the consumer group, and registers otherwise. Then it | ||
| // checks if it has subscribed for the topic, and subscribes otherwise. Later | ||
| // if a particular topic has not been consumed for | ||
| // `Config.Consumer.RegistrationTimeout` period of time, the consumer | ||
| // unsubscribes from the topic, likewise if a consumer group has not seen any | ||
| // requests for that period then the consumer deregisters from the group. | ||
| // | ||
| // implements `consumer.T`. | ||
| // implements `dispatcher.Factory`. | ||
| type t struct { | ||
| namespace *actor.ID | ||
| cfg *config.T | ||
| dispatcher *dispatcher.T | ||
| kafkaClient sarama.Client | ||
| kazooConn *kazoo.Kazoo | ||
| offsetMgrFactory offsetmgr.Factory | ||
| } | ||
|
|
||
| // Spawn creates a consumer instance with the specified configuration and | ||
| // starts all its goroutines. | ||
| func Spawn(namespace *actor.ID, cfg *config.T) (*t, error) { | ||
| saramaCfg := sarama.NewConfig() | ||
| saramaCfg.ClientID = cfg.ClientID | ||
| saramaCfg.ChannelBufferSize = cfg.Consumer.ChannelBufferSize | ||
| saramaCfg.Consumer.Offsets.CommitInterval = 50 * time.Millisecond | ||
| saramaCfg.Consumer.Retry.Backoff = cfg.Consumer.BackOffTimeout | ||
| saramaCfg.Consumer.Fetch.Default = 1024 * 1024 | ||
|
|
||
| namespace = namespace.NewChild("cons") | ||
|
|
||
| kafkaClient, err := sarama.NewClient(cfg.Kafka.SeedPeers, saramaCfg) | ||
| if err != nil { | ||
| return nil, consumer.ErrSetup(fmt.Errorf("failed to create sarama.Client: err=(%v)", err)) | ||
| } | ||
| offsetMgrFactory := offsetmgr.SpawnFactory(namespace, cfg, kafkaClient) | ||
|
|
||
| kazooCfg := kazoo.NewConfig() | ||
| kazooCfg.Chroot = cfg.ZooKeeper.Chroot | ||
| // ZooKeeper documentation says following about the session timeout: "The | ||
| // current (ZooKeeper) implementation requires that the timeout be a | ||
| // minimum of 2 times the tickTime (as set in the server configuration) and | ||
| // a maximum of 20 times the tickTime". The default tickTime is 2 seconds. | ||
| // See http://zookeeper.apache.org/doc/trunk/zookeeperProgrammers.html#ch_zkSessions | ||
| kazooCfg.Timeout = 15 * time.Second | ||
|
|
||
| kazooConn, err := kazoo.NewKazoo(cfg.ZooKeeper.SeedPeers, kazooCfg) | ||
| if err != nil { | ||
| return nil, consumer.ErrSetup(fmt.Errorf("failed to create kazoo.Kazoo: err=(%v)", err)) | ||
| } | ||
|
|
||
| c := &t{ | ||
| namespace: namespace, | ||
| cfg: cfg, | ||
| kafkaClient: kafkaClient, | ||
| offsetMgrFactory: offsetMgrFactory, | ||
| kazooConn: kazooConn, | ||
| } | ||
| c.dispatcher = dispatcher.New(c.namespace, c, c.cfg) | ||
| c.dispatcher.Start() | ||
| return c, nil | ||
| } | ||
|
|
||
| // implements `consumer.T` | ||
| func (sc *t) Consume(group, topic string) (*consumer.Message, error) { | ||
| replyCh := make(chan dispatcher.Response, 1) | ||
| sc.dispatcher.Requests() <- dispatcher.Request{time.Now().UTC(), group, topic, replyCh} | ||
| result := <-replyCh | ||
| return result.Msg, result.Err | ||
| } | ||
|
|
||
| // implements `consumer.T` | ||
| func (sc *t) Stop() { | ||
| sc.dispatcher.Stop() | ||
| } | ||
|
|
||
| // implements `dispatcher.Factory`. | ||
| func (sc *t) KeyOf(req dispatcher.Request) string { | ||
| return req.Group | ||
| } | ||
|
|
||
| // implements `dispatcher.Factory`. | ||
| func (sc *t) NewTier(key string) dispatcher.Tier { | ||
| return groupcsm.New(sc.namespace, key, sc.cfg, sc.kafkaClient, sc.kazooConn, sc.offsetMgrFactory) | ||
| } | ||
|
|
||
| // String returns a string ID of this instance to be used in logs. | ||
| func (sc *t) String() string { | ||
| return sc.namespace.String() | ||
| } |
Oops, something went wrong.