-
Notifications
You must be signed in to change notification settings - Fork 10
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
- Loading branch information
1 parent
0aa0056
commit 5d43b6a
Showing
11 changed files
with
282 additions
and
271 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
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
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,159 @@ | ||
| package kafka | ||
|
|
||
| import ( | ||
| "context" | ||
| "errors" | ||
| "fmt" | ||
| "github.com/IBM/sarama" | ||
| "reflect" | ||
| "sync" | ||
| "time" | ||
| "vc/pkg/logger" | ||
| "vc/pkg/model" | ||
| ) | ||
|
|
||
| const ( | ||
| TopicMockNext = "topic_mock_next" | ||
| TopicUpload = "topic_upload" | ||
| TypeOfStructInMessageValue = "type_of_struct_in_value" | ||
| ) | ||
|
|
||
| type HandlerConfig struct { | ||
| Topic string | ||
| ConsumerGroup string | ||
| } | ||
|
|
||
| type Consumer interface { | ||
| Start(handlerFactory func(string) sarama.ConsumerGroupHandler) error | ||
| Close(ctx context.Context) error | ||
| } | ||
|
|
||
| // MessageConsumerClient ATTENTION: Start max one instance of consumer client for each service to keep resource usage low | ||
| type MessageConsumerClient struct { | ||
| SaramaConfig *sarama.Config | ||
| brokers []string | ||
| ctx context.Context | ||
| cancel context.CancelFunc | ||
| wg sync.WaitGroup | ||
| log *logger.Log | ||
| } | ||
|
|
||
| func (c *MessageConsumerClient) CommonConsumerConfig(cfg *model.Cfg) { | ||
| //TODO(mk): set cfg from file - is now hardcoded | ||
| //TODO(mk): enable security when consumting from Kafka | ||
| c.SaramaConfig = sarama.NewConfig() | ||
| c.SaramaConfig.Consumer.Offsets.Initial = sarama.OffsetOldest | ||
| c.SaramaConfig.Consumer.Group.Rebalance.GroupStrategies = []sarama.BalanceStrategy{sarama.NewBalanceStrategyRange()} | ||
| c.SaramaConfig.Net.SASL.Enable = false | ||
| // ... | ||
| //return SaramaConfig | ||
| } | ||
|
|
||
| func NewConsumerClient(ctx context.Context, cfg *model.Cfg, brokers []string, log *logger.Log) (*MessageConsumerClient, error) { | ||
| client := &MessageConsumerClient{ | ||
| SaramaConfig: &sarama.Config{}, | ||
| brokers: brokers, | ||
| wg: sync.WaitGroup{}, | ||
| log: log, | ||
| } | ||
|
|
||
| client.CommonConsumerConfig(cfg) | ||
|
|
||
| return client, nil | ||
| } | ||
|
|
||
| func (c *MessageConsumerClient) Start(ctx context.Context, handlerFactory func(string) sarama.ConsumerGroupHandler, handlerConfigs []HandlerConfig) error { | ||
| if err := c.SaramaConfig.Validate(); err != nil { | ||
| return err | ||
| } | ||
|
|
||
| for _, handlerConfig := range handlerConfigs { | ||
| consumerGroup, err := sarama.NewConsumerGroup(c.brokers, handlerConfig.ConsumerGroup, c.SaramaConfig) | ||
| if err != nil { | ||
| c.log.Error(err, "Error creating consumer group", "group", handlerConfig.ConsumerGroup) | ||
| return err | ||
| } | ||
| c.log.Info("Started consumer group", "group", handlerConfig.ConsumerGroup) | ||
|
|
||
| var cancelCtx context.Context | ||
| cancelCtx, c.cancel = context.WithCancel(ctx) | ||
|
|
||
| c.wg.Add(1) | ||
| go func(group sarama.ConsumerGroup, topic string) { | ||
| defer c.wg.Done() | ||
| for { | ||
| handler := handlerFactory(topic) | ||
| if err := group.Consume(cancelCtx, []string{topic}, handler); err != nil { | ||
| c.log.Error(err, "Error on consumer group", "group", handlerConfig.ConsumerGroup) | ||
| //TODO(mk): use more advanced backoff algorithm? | ||
| time.Sleep(1 * time.Second) | ||
| } | ||
|
|
||
| if cancelCtx.Err() != nil { | ||
| return | ||
| } | ||
| } | ||
| }(consumerGroup, handlerConfig.Topic) | ||
| } | ||
| return nil | ||
| } | ||
|
|
||
| func (c *MessageConsumerClient) Close(ctx context.Context) error { | ||
| c.cancel() | ||
| c.wg.Wait() | ||
| c.log.Info("Closed") | ||
| return nil | ||
| } | ||
|
|
||
| type MessageHandler interface { | ||
| HandleMessage(ctx context.Context, message *sarama.ConsumerMessage) error | ||
| } | ||
|
|
||
| type ConsumerGroupHandler struct { | ||
| Handlers map[string]MessageHandler | ||
| Log *logger.Log | ||
| } | ||
|
|
||
| func (cgh *ConsumerGroupHandler) Setup(_ sarama.ConsumerGroupSession) error { return nil } | ||
| func (cgh *ConsumerGroupHandler) Cleanup(_ sarama.ConsumerGroupSession) error { return nil } | ||
|
|
||
| func (cgh *ConsumerGroupHandler) ConsumeClaim(session sarama.ConsumerGroupSession, claim sarama.ConsumerGroupClaim) error { | ||
| if cgh.Handlers == nil { | ||
| cgh.Log.Error(errors.New("No handlers defined"), "No Handlers for any topic") | ||
| //TODO(mk): send to a general error topic? | ||
| return nil | ||
| } | ||
|
|
||
| handler, exists := cgh.Handlers[claim.Topic()] | ||
| if !exists { | ||
| cgh.Log.Error(errors.New("No handler for topic"), "topic", claim.Topic()) | ||
| //TODO(mk): send to a general error topic? | ||
| return nil | ||
| } | ||
|
|
||
| handlerType := reflect.TypeOf(handler).String() | ||
|
|
||
| for message := range claim.Messages() { | ||
| var errMessage string | ||
|
|
||
| if err := handler.HandleMessage(session.Context(), message); err != nil { | ||
| cgh.Log.Error(err, "Error handling message", "topic", claim.Topic()) | ||
| //TODO(mk): more advanced retry/error handling including send to error topic if not OK after X number of retries | ||
| errMessage = fmt.Sprintf("error handling message: %v", err) | ||
| } | ||
|
|
||
| info := fmt.Sprintf("message consumed by handler type: %s, topic: %s, partition: %d, offset: %d", | ||
| handlerType, | ||
| claim.Topic(), | ||
| message.Partition, | ||
| message.Offset, | ||
| ) | ||
| if errMessage != "" { | ||
| info = fmt.Sprintf("%s, error: %s", info, errMessage) | ||
| } | ||
| cgh.Log.Debug("Consumed message", "info", info) | ||
|
|
||
| session.MarkMessage(message, info) | ||
| } | ||
| return nil | ||
| } |
Oops, something went wrong.