dep ensure，解决依赖问题

Gopkg.lock

@@ -1280,11 +1280,11 @@
 
 [[projects]]
   branch = "master"
-  digest = "1:e52c4c426ba5bbb92b2d6b2f20f694ff836dde16151adaca39c86d5fb010fea9"
+  digest = "1:a22a70d0156f2b836993f4d07b38c4b1c2058d61ced4e55bb7fb77b297424a04"
   name = "yunion.io/x/sqlchemy"
   packages = ["."]
   pruneopts = "UT"
-  revision = "946ed13318f85073e55dbb8105593c710e943a31"
+  revision = "3342ff524267b69938600968f3f193fc794b309c"
 
 [[projects]]
   branch = "master"

vendor/github.com/360EntSecGroup-Skylar/excelize/PULL_REQUEST_TEMPLATE.md

@@ -0,0 +1,45 @@
+# PR Details
+
+<!--- Provide a general summary of your changes in the Title above -->
+
+## Description
+
+<!--- Describe your changes in detail -->
+
+## Related Issue
+
+<!--- This project only accepts pull requests related to open issues -->
+<!--- If suggesting a new feature or change, please discuss it in an issue first -->
+<!--- If fixing a bug, there should be an issue describing it with steps to reproduce -->
+<!--- Please link to the issue here: -->
+
+## Motivation and Context
+
+<!--- Why is this change required? What problem does it solve? -->
+
+## How Has This Been Tested
+
+<!--- Please describe in detail how you tested your changes. -->
+<!--- Include details of your testing environment, and the tests you ran to -->
+<!--- see how your change affects other areas of the code, etc. -->
+
+## Types of changes
+
+<!--- What types of changes does your code introduce? Put an `x` in all the boxes that apply: -->
+
+- [ ] Docs change / refactoring / dependency upgrade
+- [ ] Bug fix (non-breaking change which fixes an issue)
+- [ ] New feature (non-breaking change which adds functionality)
+- [ ] Breaking change (fix or feature that would cause existing functionality to change)
+
+## Checklist
+
+<!--- Go over all the following points, and put an `x` in all the boxes that apply. -->
+<!--- If you're unsure about any of these, don't hesitate to ask. We're here to help! -->
+
+- [ ] My code follows the code style of this project.
+- [ ] My change requires a change to the documentation.
+- [ ] I have updated the documentation accordingly.
+- [ ] I have read the **CONTRIBUTING** document.
+- [ ] I have added tests to cover my changes.
+- [ ] All new and existing tests passed.

vendor/github.com/360EntSecGroup-Skylar/excelize/datavalidation.go

@@ -0,0 +1,233 @@
+// Copyright 2016 - 2018 The excelize Authors. All rights reserved. Use of
+// this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+//
+// Package excelize providing a set of functions that allow you to write to
+// and read from XLSX files. Support reads and writes XLSX file generated by
+// Microsoft Excel™ 2007 and later. Support save file without losing original
+// charts of XLSX. This library needs Go version 1.8 or later.
+
+package excelize
+
+import (
+	"fmt"
+	"strings"
+)
+
+// DataValidationType defined the type of data validation.
+type DataValidationType int
+
+// Data validation types.
+const (
+	_DataValidationType = iota
+	typeNone            // inline use
+	DataValidationTypeCustom
+	DataValidationTypeDate
+	DataValidationTypeDecimal
+	typeList // inline use
+	DataValidationTypeTextLeng
+	DataValidationTypeTime
+	// DataValidationTypeWhole Integer
+	DataValidationTypeWhole
+)
+
+const (
+	// dataValidationFormulaStrLen 255 characters+ 2 quotes
+	dataValidationFormulaStrLen = 257
+	// dataValidationFormulaStrLenErr
+	dataValidationFormulaStrLenErr = "data validation must be 0-255 characters"
+)
+
+// DataValidationErrorStyle defined the style of data validation error alert.
+type DataValidationErrorStyle int
+
+// Data validation error styles.
+const (
+	_ DataValidationErrorStyle = iota
+	DataValidationErrorStyleStop
+	DataValidationErrorStyleWarning
+	DataValidationErrorStyleInformation
+)
+
+// Data validation error styles.
+const (
+	styleStop        = "stop"
+	styleWarning     = "warning"
+	styleInformation = "information"
+)
+
+// DataValidationOperator operator enum.
+type DataValidationOperator int
+
+// Data validation operators.
+const (
+	_DataValidationOperator = iota
+	DataValidationOperatorBetween
+	DataValidationOperatorEqual
+	DataValidationOperatorGreaterThan
+	DataValidationOperatorGreaterThanOrEqual
+	DataValidationOperatorLessThan
+	DataValidationOperatorLessThanOrEqual
+	DataValidationOperatorNotBetween
+	DataValidationOperatorNotEqual
+)
+
+// NewDataValidation return data validation struct.
+func NewDataValidation(allowBlank bool) *DataValidation {
+	return &DataValidation{
+		AllowBlank:       allowBlank,
+		ShowErrorMessage: false,
+		ShowInputMessage: false,
+	}
+}
+
+// SetError set error notice.
+func (dd *DataValidation) SetError(style DataValidationErrorStyle, title, msg string) {
+	dd.Error = &msg
+	dd.ErrorTitle = &title
+	strStyle := styleStop
+	switch style {
+	case DataValidationErrorStyleStop:
+		strStyle = styleStop
+	case DataValidationErrorStyleWarning:
+		strStyle = styleWarning
+	case DataValidationErrorStyleInformation:
+		strStyle = styleInformation
+
+	}
+	dd.ShowErrorMessage = true
+	dd.ErrorStyle = &strStyle
+}
+
+// SetInput set prompt notice.
+func (dd *DataValidation) SetInput(title, msg string) {
+	dd.ShowInputMessage = true
+	dd.PromptTitle = &title
+	dd.Prompt = &msg
+}
+
+// SetDropList data validation list.
+func (dd *DataValidation) SetDropList(keys []string) error {
+	dd.Formula1 = "\"" + strings.Join(keys, ",") + "\""
+	dd.Type = convDataValidationType(typeList)
+	return nil
+}
+
+// SetRange provides function to set data validation range in drop list.
+func (dd *DataValidation) SetRange(f1, f2 int, t DataValidationType, o DataValidationOperator) error {
+	formula1 := fmt.Sprintf("%d", f1)
+	formula2 := fmt.Sprintf("%d", f2)
+	if dataValidationFormulaStrLen < len(dd.Formula1) || dataValidationFormulaStrLen < len(dd.Formula2) {
+		return fmt.Errorf(dataValidationFormulaStrLenErr)
+	}
+
+	dd.Formula1 = formula1
+	dd.Formula2 = formula2
+	dd.Type = convDataValidationType(t)
+	dd.Operator = convDataValidationOperatior(o)
+	return nil
+}
+
+// SetSqrefDropList provides set data validation on a range with source
+// reference range of the worksheet by given data validation object and
+// worksheet name. The data validation object can be created by
+// NewDataValidation function. For example, set data validation on
+// Sheet1!A7:B8 with validation criteria source Sheet1!E1:E3 settings, create
+// in-cell dropdown by allowing list source:
+//
+//     dvRange := excelize.NewDataValidation(true)
+//     dvRange.Sqref = "A7:B8"
+//     dvRange.SetSqrefDropList("E1:E3", true)
+//     xlsx.AddDataValidation("Sheet1", dvRange)
+//
+func (dd *DataValidation) SetSqrefDropList(sqref string, isCurrentSheet bool) error {
+	if isCurrentSheet {
+		dd.Formula1 = sqref
+		dd.Type = convDataValidationType(typeList)
+		return nil
+	}
+	return fmt.Errorf("cross-sheet sqref cell are not supported")
+}
+
+// SetSqref provides function to set data validation range in drop list.
+func (dd *DataValidation) SetSqref(sqref string) {
+	if dd.Sqref == "" {
+		dd.Sqref = sqref
+	} else {
+		dd.Sqref = fmt.Sprintf("%s %s", dd.Sqref, sqref)
+	}
+}
+
+// convDataValidationType get excel data validation type.
+func convDataValidationType(t DataValidationType) string {
+	typeMap := map[DataValidationType]string{
+		typeNone:                   "none",
+		DataValidationTypeCustom:   "custom",
+		DataValidationTypeDate:     "date",
+		DataValidationTypeDecimal:  "decimal",
+		typeList:                   "list",
+		DataValidationTypeTextLeng: "textLength",
+		DataValidationTypeTime:     "time",
+		DataValidationTypeWhole:    "whole",
+	}
+
+	return typeMap[t]
+
+}
+
+// convDataValidationOperatior get excel data validation operator.
+func convDataValidationOperatior(o DataValidationOperator) string {
+	typeMap := map[DataValidationOperator]string{
+		DataValidationOperatorBetween:            "between",
+		DataValidationOperatorEqual:              "equal",
+		DataValidationOperatorGreaterThan:        "greaterThan",
+		DataValidationOperatorGreaterThanOrEqual: "greaterThanOrEqual",
+		DataValidationOperatorLessThan:           "lessThan",
+		DataValidationOperatorLessThanOrEqual:    "lessThanOrEqual",
+		DataValidationOperatorNotBetween:         "notBetween",
+		DataValidationOperatorNotEqual:           "notEqual",
+	}
+
+	return typeMap[o]
+
+}
+
+// AddDataValidation provides set data validation on a range of the worksheet
+// by given data validation object and worksheet name. The data validation
+// object can be created by NewDataValidation function.
+//
+// Example 1, set data validation on Sheet1!A1:B2 with validation criteria
+// settings, show error alert after invalid data is entered with "Stop" style
+// and custom title "error body":
+//
+//     dvRange := excelize.NewDataValidation(true)
+//     dvRange.Sqref = "A1:B2"
+//     dvRange.SetRange(10, 20, excelize.DataValidationTypeWhole, excelize.DataValidationOperatorBetween)
+//     dvRange.SetError(excelize.DataValidationErrorStyleStop, "error title", "error body")
+//     xlsx.AddDataValidation("Sheet1", dvRange)
+//
+// Example 2, set data validation on Sheet1!A3:B4 with validation criteria
+// settings, and show input message when cell is selected:
+//
+//     dvRange = excelize.NewDataValidation(true)
+//     dvRange.Sqref = "A3:B4"
+//     dvRange.SetRange(10, 20, excelize.DataValidationTypeWhole, excelize.DataValidationOperatorGreaterThan)
+//     dvRange.SetInput("input title", "input body")
+//     xlsx.AddDataValidation("Sheet1", dvRange)
+//
+// Example 3, set data validation on Sheet1!A5:B6 with validation criteria
+// settings, create in-cell dropdown by allowing list source:
+//
+//     dvRange = excelize.NewDataValidation(true)
+//     dvRange.Sqref = "A5:B6"
+//     dvRange.SetDropList([]string{"1", "2", "3"})
+//     xlsx.AddDataValidation("Sheet1", dvRange)
+//
+func (f *File) AddDataValidation(sheet string, dv *DataValidation) {
+	xlsx := f.workSheetReader(sheet)
+	if nil == xlsx.DataValidations {
+		xlsx.DataValidations = new(xlsxDataValidations)
+	}
+	xlsx.DataValidations.DataValidation = append(xlsx.DataValidations.DataValidation, dv)
+	xlsx.DataValidations.Count = len(xlsx.DataValidations.DataValidation)
+}

vendor/github.com/360EntSecGroup-Skylar/excelize/hsl.go

@@ -0,0 +1,139 @@
+// Copyright (c) 2012 Rodrigo Moraes. All rights reserved.
+//
+// Redistribution and use in source and binary forms, with or without
+// modification, are permitted provided that the following conditions are
+// met:
+//
+// 	 * Redistributions of source code must retain the above copyright
+// notice, this list of conditions and the following disclaimer.
+// 	 * Redistributions in binary form must reproduce the above
+// copyright notice, this list of conditions and the following disclaimer
+// in the documentation and/or other materials provided with the
+// distribution.
+// 	 * Neither the name of Google Inc. nor the names of its
+// contributors may be used to endorse or promote products derived from
+// this software without specific prior written permission.
+//
+// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS
+// "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT
+// LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR
+// A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT
+// OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL,
+// SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT
+// LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE,
+// DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY
+// THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+// (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
+// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+
+package excelize
+
+import (
+	"image/color"
+	"math"
+)
+
+// HSLModel converts any color.Color to a HSL color.
+var HSLModel = color.ModelFunc(hslModel)
+
+// HSL represents a cylindrical coordinate of points in an RGB color model.
+//
+// Values are in the range 0 to 1.
+type HSL struct {
+	H, S, L float64
+}
+
+// RGBA returns the alpha-premultiplied red, green, blue and alpha values
+// for the HSL.
+func (c HSL) RGBA() (uint32, uint32, uint32, uint32) {
+	r, g, b := HSLToRGB(c.H, c.S, c.L)
+	return uint32(r) * 0x101, uint32(g) * 0x101, uint32(b) * 0x101, 0xffff
+}
+
+// hslModel converts a color.Color to HSL.
+func hslModel(c color.Color) color.Color {
+	if _, ok := c.(HSL); ok {
+		return c
+	}
+	r, g, b, _ := c.RGBA()
+	h, s, l := RGBToHSL(uint8(r>>8), uint8(g>>8), uint8(b>>8))
+	return HSL{h, s, l}
+}
+
+// RGBToHSL converts an RGB triple to a HSL triple.
+func RGBToHSL(r, g, b uint8) (h, s, l float64) {
+	fR := float64(r) / 255
+	fG := float64(g) / 255
+	fB := float64(b) / 255
+	max := math.Max(math.Max(fR, fG), fB)
+	min := math.Min(math.Min(fR, fG), fB)
+	l = (max + min) / 2
+	if max == min {
+		// Achromatic.
+		h, s = 0, 0
+	} else {
+		// Chromatic.
+		d := max - min
+		if l > 0.5 {
+			s = d / (2.0 - max - min)
+		} else {
+			s = d / (max + min)
+		}
+		switch max {
+		case fR:
+			h = (fG - fB) / d
+			if fG < fB {
+				h += 6
+			}
+		case fG:
+			h = (fB-fR)/d + 2
+		case fB:
+			h = (fR-fG)/d + 4
+		}
+		h /= 6
+	}
+	return
+}
+
+// HSLToRGB converts an HSL triple to a RGB triple.
+func HSLToRGB(h, s, l float64) (r, g, b uint8) {
+	var fR, fG, fB float64
+	if s == 0 {
+		fR, fG, fB = l, l, l
+	} else {
+		var q float64
+		if l < 0.5 {
+			q = l * (1 + s)
+		} else {
+			q = l + s - s*l
+		}
+		p := 2*l - q
+		fR = hueToRGB(p, q, h+1.0/3)
+		fG = hueToRGB(p, q, h)
+		fB = hueToRGB(p, q, h-1.0/3)
+	}
+	r = uint8((fR * 255) + 0.5)
+	g = uint8((fG * 255) + 0.5)
+	b = uint8((fB * 255) + 0.5)
+	return
+}
+
+// hueToRGB is a helper function for HSLToRGB.
+func hueToRGB(p, q, t float64) float64 {
+	if t < 0 {
+		t++
+	}
+	if t > 1 {
+		t--
+	}
+	if t < 1.0/6 {
+		return p + (q-p)*6*t
+	}
+	if t < 0.5 {
+		return q
+	}
+	if t < 2.0/3 {
+		return p + (q-p)*(2.0/3-t)*6
+	}
+	return p
+}

vendor/github.com/360EntSecGroup-Skylar/excelize/xmlTheme.go

@@ -0,0 +1,149 @@
+// Copyright 2016 - 2018 The excelize Authors. All rights reserved. Use of
+// this source code is governed by a BSD-style license that can be found in
+// the LICENSE file.
+//
+// Package excelize providing a set of functions that allow you to write to
+// and read from XLSX files. Support reads and writes XLSX file generated by
+// Microsoft Excel™ 2007 and later. Support save file without losing original
+// charts of XLSX. This library needs Go version 1.8 or later.
+
+package excelize
+
+import "encoding/xml"
+
+// xlsxTheme directly maps the theme element in the namespace
+// http://schemas.openxmlformats.org/drawingml/2006/main
+type xlsxTheme struct {
+	ThemeElements     xlsxThemeElements     `xml:"themeElements"`
+	ObjectDefaults    xlsxObjectDefaults    `xml:"objectDefaults"`
+	ExtraClrSchemeLst xlsxExtraClrSchemeLst `xml:"extraClrSchemeLst"`
+	ExtLst            *xlsxExtLst           `xml:"extLst"`
+}
+
+// objectDefaults element allows for the definition of default shape, line,
+// and textbox formatting properties. An application can use this information
+// to format a shape (or text) initially on insertion into a document.
+type xlsxObjectDefaults struct {
+	ObjectDefaults string `xml:",innerxml"`
+}
+
+// xlsxExtraClrSchemeLst element is a container for the list of extra color
+// schemes present in a document.
+type xlsxExtraClrSchemeLst struct {
+	ExtraClrSchemeLst string `xml:",innerxml"`
+}
+
+// xlsxThemeElements directly maps the element defines the theme formatting
+// options for the theme and is the workhorse of the theme. This is where the
+// bulk of the shared theme information is contained and used by a document.
+// This element contains the color scheme, font scheme, and format scheme
+// elements which define the different formatting aspects of what a theme
+// defines.
+type xlsxThemeElements struct {
+	ClrScheme  xlsxClrScheme  `xml:"clrScheme"`
+	FontScheme xlsxFontScheme `xml:"fontScheme"`
+	FmtScheme  xlsxFmtScheme  `xml:"fmtScheme"`
+}
+
+// xlsxClrScheme element specifies the theme color, stored in the document's
+// Theme part to which the value of this theme color shall be mapped. This
+// mapping enables multiple theme colors to be chained together.
+type xlsxClrScheme struct {
+	Name     string            `xml:"name,attr"`
+	Children []xlsxClrSchemeEl `xml:",any"`
+}
+
+// xlsxFontScheme element defines the font scheme within the theme. The font
+// scheme consists of a pair of major and minor fonts for which to use in a
+// document. The major font corresponds well with the heading areas of a
+// document, and the minor font corresponds well with the normal text or
+// paragraph areas.
+type xlsxFontScheme struct {
+	Name      string        `xml:"name,attr"`
+	MajorFont xlsxMajorFont `xml:"majorFont"`
+	MinorFont xlsxMinorFont `xml:"minorFont"`
+	ExtLst    *xlsxExtLst   `xml:"extLst"`
+}
+
+// xlsxMajorFont element defines the set of major fonts which are to be used
+// under different languages or locals.
+type xlsxMajorFont struct {
+	Children []xlsxFontSchemeEl `xml:",any"`
+}
+
+// xlsxMinorFont element defines the set of minor fonts that are to be used
+// under different languages or locals.
+type xlsxMinorFont struct {
+	Children []xlsxFontSchemeEl `xml:",any"`
+}
+
+// xlsxFmtScheme element contains the background fill styles, effect styles,
+// fill styles, and line styles which define the style matrix for a theme. The
+// style matrix consists of subtle, moderate, and intense fills, lines, and
+// effects. The background fills are not generally thought of to directly be
+// associated with the matrix, but do play a role in the style of the overall
+// document. Usually, a given object chooses a single line style, a single
+// fill style, and a single effect style in order to define the overall final
+// look of the object.
+type xlsxFmtScheme struct {
+	Name           string             `xml:"name,attr"`
+	FillStyleLst   xlsxFillStyleLst   `xml:"fillStyleLst"`
+	LnStyleLst     xlsxLnStyleLst     `xml:"lnStyleLst"`
+	EffectStyleLst xlsxEffectStyleLst `xml:"effectStyleLst"`
+	BgFillStyleLst xlsxBgFillStyleLst `xml:"bgFillStyleLst"`
+}
+
+// xlsxFillStyleLst element defines a set of three fill styles that are used
+// within a theme. The three fill styles are arranged in order from subtle to
+// moderate to intense.
+type xlsxFillStyleLst struct {
+	FillStyleLst string `xml:",innerxml"`
+}
+
+// xlsxLnStyleLst element defines a list of three line styles for use within a
+// theme. The three line styles are arranged in order from subtle to moderate
+// to intense versions of lines. This list makes up part of the style matrix.
+type xlsxLnStyleLst struct {
+	LnStyleLst string `xml:",innerxml"`
+}
+
+// xlsxEffectStyleLst element defines a set of three effect styles that create
+// the effect style list for a theme. The effect styles are arranged in order
+// of subtle to moderate to intense.
+type xlsxEffectStyleLst struct {
+	EffectStyleLst string `xml:",innerxml"`
+}
+
+// xlsxBgFillStyleLst  element defines a list of background fills that are
+// used within a theme. The background fills consist of three fills, arranged
+// in order from subtle to moderate to intense.
+type xlsxBgFillStyleLst struct {
+	BgFillStyleLst string `xml:",innerxml"`
+}
+
+// xlsxClrScheme maps to children of the clrScheme element in the namespace
+// http://schemas.openxmlformats.org/drawingml/2006/main - currently I have
+// not checked it for completeness - it does as much as I need.
+type xlsxClrSchemeEl struct {
+	XMLName xml.Name
+	SysClr  *xlsxSysClr    `xml:"sysClr"`
+	SrgbClr *attrValString `xml:"srgbClr"`
+}
+
+// xlsxFontSchemeEl directly maps the major and minor font of the style's font
+// scheme.
+type xlsxFontSchemeEl struct {
+	XMLName     xml.Name
+	Script      string `xml:"script,attr,omitempty"`
+	Typeface    string `xml:"typeface,attr"`
+	Panose      string `xml:"panose,attr,omitempty"`
+	PitchFamily string `xml:"pitchFamily,attr,omitempty"`
+	Charset     string `xml:"charset,attr,omitempty"`
+}
+
+// xlsxSysClr element specifies a color bound to predefined operating system
+// elements.
+type xlsxSysClr struct {
+	Val     string `xml:"val,attr"`
+	LastClr string `xml:"lastClr,attr"`
+}

vendor/github.com/Shopify/sarama/balance_strategy.go

@@ -0,0 +1,129 @@
+package sarama
+
+import (
+	"math"
+	"sort"
+)
+
+// BalanceStrategyPlan is the results of any BalanceStrategy.Plan attempt.
+// It contains an allocation of topic/partitions by memberID in the form of
+// a `memberID -> topic -> partitions` map.
+type BalanceStrategyPlan map[string]map[string][]int32
+
+// Add assigns a topic with a number partitions to a member.
+func (p BalanceStrategyPlan) Add(memberID, topic string, partitions ...int32) {
+	if len(partitions) == 0 {
+		return
+	}
+	if _, ok := p[memberID]; !ok {
+		p[memberID] = make(map[string][]int32, 1)
+	}
+	p[memberID][topic] = append(p[memberID][topic], partitions...)
+}
+
+// --------------------------------------------------------------------
+
+// BalanceStrategy is used to balance topics and partitions
+// across memebers of a consumer group
+type BalanceStrategy interface {
+	// Name uniquely identifies the strategy.
+	Name() string
+
+	// Plan accepts a map of `memberID -> metadata` and a map of `topic -> partitions`
+	// and returns a distribution plan.
+	Plan(members map[string]ConsumerGroupMemberMetadata, topics map[string][]int32) (BalanceStrategyPlan, error)
+}
+
+// --------------------------------------------------------------------
+
+// BalanceStrategyRange is the default and assigns partitions as ranges to consumer group members.
+// Example with one topic T with six partitions (0..5) and two members (M1, M2):
+//   M1: {T: [0, 1, 2]}
+//   M2: {T: [3, 4, 5]}
+var BalanceStrategyRange = &balanceStrategy{
+	name: "range",
+	coreFn: func(plan BalanceStrategyPlan, memberIDs []string, topic string, partitions []int32) {
+		step := float64(len(partitions)) / float64(len(memberIDs))
+
+		for i, memberID := range memberIDs {
+			pos := float64(i)
+			min := int(math.Floor(pos*step + 0.5))
+			max := int(math.Floor((pos+1)*step + 0.5))
+			plan.Add(memberID, topic, partitions[min:max]...)
+		}
+	},
+}
+
+// BalanceStrategyRoundRobin assigns partitions to members in alternating order.
+// Example with topic T with six partitions (0..5) and two members (M1, M2):
+//   M1: {T: [0, 2, 4]}
+//   M2: {T: [1, 3, 5]}
+var BalanceStrategyRoundRobin = &balanceStrategy{
+	name: "roundrobin",
+	coreFn: func(plan BalanceStrategyPlan, memberIDs []string, topic string, partitions []int32) {
+		for i, part := range partitions {
+			memberID := memberIDs[i%len(memberIDs)]
+			plan.Add(memberID, topic, part)
+		}
+	},
+}
+
+// --------------------------------------------------------------------
+
+type balanceStrategy struct {
+	name   string
+	coreFn func(plan BalanceStrategyPlan, memberIDs []string, topic string, partitions []int32)
+}
+
+// Name implements BalanceStrategy.
+func (s *balanceStrategy) Name() string { return s.name }
+
+// Balance implements BalanceStrategy.
+func (s *balanceStrategy) Plan(members map[string]ConsumerGroupMemberMetadata, topics map[string][]int32) (BalanceStrategyPlan, error) {
+	// Build members by topic map
+	mbt := make(map[string][]string)
+	for memberID, meta := range members {
+		for _, topic := range meta.Topics {
+			mbt[topic] = append(mbt[topic], memberID)
+		}
+	}
+
+	// Sort members for each topic
+	for topic, memberIDs := range mbt {
+		sort.Sort(&balanceStrategySortable{
+			topic:     topic,
+			memberIDs: memberIDs,
+		})
+	}
+
+	// Assemble plan
+	plan := make(BalanceStrategyPlan, len(members))
+	for topic, memberIDs := range mbt {
+		s.coreFn(plan, memberIDs, topic, topics[topic])
+	}
+	return plan, nil
+}
+
+type balanceStrategySortable struct {
+	topic     string
+	memberIDs []string
+}
+
+func (p balanceStrategySortable) Len() int { return len(p.memberIDs) }
+func (p balanceStrategySortable) Swap(i, j int) {
+	p.memberIDs[i], p.memberIDs[j] = p.memberIDs[j], p.memberIDs[i]
+}
+func (p balanceStrategySortable) Less(i, j int) bool {
+	return balanceStrategyHashValue(p.topic, p.memberIDs[i]) < balanceStrategyHashValue(p.topic, p.memberIDs[j])
+}
+
+func balanceStrategyHashValue(vv ...string) uint32 {
+	h := uint32(2166136261)
+	for _, s := range vv {
+		for _, c := range s {
+			h ^= uint32(c)
+			h *= 16777619
+		}
+	}
+	return h
+}

vendor/github.com/Shopify/sarama/consumer_group.go

@@ -0,0 +1,774 @@
+package sarama
+
+import (
+	"context"
+	"errors"
+	"fmt"
+	"sort"
+	"sync"
+	"time"
+)
+
+// ErrClosedConsumerGroup is the error returned when a method is called on a consumer group that has been closed.
+var ErrClosedConsumerGroup = errors.New("kafka: tried to use a consumer group that was closed")
+
+// ConsumerGroup is responsible for dividing up processing of topics and partitions
+// over a collection of processes (the members of the consumer group).
+type ConsumerGroup interface {
+	// Consume joins a cluster of consumers for a given list of topics and
+	// starts a blocking ConsumerGroupSession through the ConsumerGroupHandler.
+	//
+	// The life-cycle of a session is represented by the following steps:
+	//
+	// 1. The consumers join the group (as explained in https://kafka.apache.org/documentation/#intro_consumers)
+	//    and is assigned their "fair share" of partitions, aka 'claims'.
+	// 2. Before processing starts, the handler's Setup() hook is called to notify the user
+	//    of the claims and allow any necessary preparation or alteration of state.
+	// 3. For each of the assigned claims the handler's ConsumeClaim() function is then called
+	//    in a separate goroutine which requires it to be thread-safe. Any state must be carefully protected
+	//    from concurrent reads/writes.
+	// 4. The session will persist until one of the ConsumeClaim() functions exits. This can be either when the
+	//    parent context is cancelled or when a server-side rebalance cycle is initiated.
+	// 5. Once all the ConsumeClaim() loops have exited, the handler's Cleanup() hook is called
+	//    to allow the user to perform any final tasks before a rebalance.
+	// 6. Finally, marked offsets are committed one last time before claims are released.
+	//
+	// Please note, that once a relance is triggered, sessions must be completed within
+	// Config.Consumer.Group.Rebalance.Timeout. This means that ConsumeClaim() functions must exit
+	// as quickly as possible to allow time for Cleanup() and the final offset commit. If the timeout
+	// is exceeded, the consumer will be removed from the group by Kafka, which will cause offset
+	// commit failures.
+	Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error
+
+	// Errors returns a read channel of errors that occurred during the consumer life-cycle.
+	// By default, errors are logged and not returned over this channel.
+	// If you want to implement any custom error handling, set your config's
+	// Consumer.Return.Errors setting to true, and read from this channel.
+	Errors() <-chan error
+
+	// Close stops the ConsumerGroup and detaches any running sessions. It is required to call
+	// this function before the object passes out of scope, as it will otherwise leak memory.
+	Close() error
+}
+
+type consumerGroup struct {
+	client    Client
+	ownClient bool
+
+	config   *Config
+	consumer Consumer
+	groupID  string
+	memberID string
+	errors   chan error
+
+	lock      sync.Mutex
+	closed    chan none
+	closeOnce sync.Once
+}
+
+// NewConsumerGroup creates a new consumer group the given broker addresses and configuration.
+func NewConsumerGroup(addrs []string, groupID string, config *Config) (ConsumerGroup, error) {
+	client, err := NewClient(addrs, config)
+	if err != nil {
+		return nil, err
+	}
+
+	c, err := NewConsumerGroupFromClient(groupID, client)
+	if err != nil {
+		_ = client.Close()
+		return nil, err
+	}
+
+	c.(*consumerGroup).ownClient = true
+	return c, nil
+}
+
+// NewConsumerFromClient creates a new consumer group using the given client. It is still
+// necessary to call Close() on the underlying client when shutting down this consumer.
+// PLEASE NOTE: consumer groups can only re-use but not share clients.
+func NewConsumerGroupFromClient(groupID string, client Client) (ConsumerGroup, error) {
+	config := client.Config()
+	if !config.Version.IsAtLeast(V0_10_2_0) {
+		return nil, ConfigurationError("consumer groups require Version to be >= V0_10_2_0")
+	}
+
+	consumer, err := NewConsumerFromClient(client)
+	if err != nil {
+		return nil, err
+	}
+
+	return &consumerGroup{
+		client:   client,
+		consumer: consumer,
+		config:   config,
+		groupID:  groupID,
+		errors:   make(chan error, config.ChannelBufferSize),
+		closed:   make(chan none),
+	}, nil
+}
+
+// Errors implements ConsumerGroup.
+func (c *consumerGroup) Errors() <-chan error { return c.errors }
+
+// Close implements ConsumerGroup.
+func (c *consumerGroup) Close() (err error) {
+	c.closeOnce.Do(func() {
+		close(c.closed)
+
+		c.lock.Lock()
+		defer c.lock.Unlock()
+
+		// leave group
+		if e := c.leave(); e != nil {
+			err = e
+		}
+
+		// drain errors
+		go func() {
+			close(c.errors)
+		}()
+		for e := range c.errors {
+			err = e
+		}
+
+		if c.ownClient {
+			if e := c.client.Close(); e != nil {
+				err = e
+			}
+		}
+	})
+	return
+}
+
+// Consume implements ConsumerGroup.
+func (c *consumerGroup) Consume(ctx context.Context, topics []string, handler ConsumerGroupHandler) error {
+	// Ensure group is not closed
+	select {
+	case <-c.closed:
+		return ErrClosedConsumerGroup
+	default:
+	}
+
+	c.lock.Lock()
+	defer c.lock.Unlock()
+
+	// Quick exit when no topics are provided
+	if len(topics) == 0 {
+		return fmt.Errorf("no topics provided")
+	}
+
+	// Refresh metadata for requested topics
+	if err := c.client.RefreshMetadata(topics...); err != nil {
+		return err
+	}
+
+	// Get coordinator
+	coordinator, err := c.client.Coordinator(c.groupID)
+	if err != nil {
+		return err
+	}
+
+	// Init session
+	sess, err := c.newSession(ctx, coordinator, topics, handler, c.config.Consumer.Group.Rebalance.Retry.Max)
+	if err == ErrClosedClient {
+		return ErrClosedConsumerGroup
+	} else if err != nil {
+		return err
+	}
+
+	// Wait for session exit signal
+	<-sess.ctx.Done()
+
+	// Gracefully release session claims
+	return sess.release(true)
+}
+
+func (c *consumerGroup) newSession(ctx context.Context, coordinator *Broker, topics []string, handler ConsumerGroupHandler, retries int) (*consumerGroupSession, error) {
+	// Join consumer group
+	join, err := c.joinGroupRequest(coordinator, topics)
+	if err != nil {
+		_ = coordinator.Close()
+		return nil, err
+	}
+	switch join.Err {
+	case ErrNoError:
+		c.memberID = join.MemberId
+	case ErrUnknownMemberId, ErrIllegalGeneration: // reset member ID and retry immediately
+		c.memberID = ""
+		return c.newSession(ctx, coordinator, topics, handler, retries)
+	case ErrRebalanceInProgress: // retry after backoff
+		if retries <= 0 {
+			return nil, join.Err
+		}
+
+		select {
+		case <-c.closed:
+			return nil, ErrClosedConsumerGroup
+		case <-time.After(c.config.Consumer.Group.Rebalance.Retry.Backoff):
+		}
+
+		return c.newSession(ctx, coordinator, topics, handler, retries-1)
+	default:
+		return nil, join.Err
+	}
+
+	// Prepare distribution plan if we joined as the leader
+	var plan BalanceStrategyPlan
+	if join.LeaderId == join.MemberId {
+		members, err := join.GetMembers()
+		if err != nil {
+			return nil, err
+		}
+
+		plan, err = c.balance(members)
+		if err != nil {
+			return nil, err
+		}
+	}
+
+	// Sync consumer group
+	sync, err := c.syncGroupRequest(coordinator, plan, join.GenerationId)
+	if err != nil {
+		_ = coordinator.Close()
+		return nil, err
+	}
+	switch sync.Err {
+	case ErrNoError:
+	case ErrUnknownMemberId, ErrIllegalGeneration: // reset member ID and retry immediately
+		c.memberID = ""
+		return c.newSession(ctx, coordinator, topics, handler, retries)
+	case ErrRebalanceInProgress: // retry after backoff
+		if retries <= 0 {
+			return nil, sync.Err
+		}
+
+		select {
+		case <-c.closed:
+			return nil, ErrClosedConsumerGroup
+		case <-time.After(c.config.Consumer.Group.Rebalance.Retry.Backoff):
+		}
+
+		return c.newSession(ctx, coordinator, topics, handler, retries-1)
+	default:
+		return nil, sync.Err
+	}
+
+	// Retrieve and sort claims
+	var claims map[string][]int32
+	if len(sync.MemberAssignment) > 0 {
+		members, err := sync.GetMemberAssignment()
+		if err != nil {
+			return nil, err
+		}
+		claims = members.Topics
+
+		for _, partitions := range claims {
+			sort.Sort(int32Slice(partitions))
+		}
+	}
+
+	return newConsumerGroupSession(c, ctx, claims, join.MemberId, join.GenerationId, handler)
+}
+
+func (c *consumerGroup) joinGroupRequest(coordinator *Broker, topics []string) (*JoinGroupResponse, error) {
+	req := &JoinGroupRequest{
+		GroupId:        c.groupID,
+		MemberId:       c.memberID,
+		SessionTimeout: int32(c.config.Consumer.Group.Session.Timeout / time.Millisecond),
+		ProtocolType:   "consumer",
+	}
+	if c.config.Version.IsAtLeast(V0_10_1_0) {
+		req.Version = 1
+		req.RebalanceTimeout = int32(c.config.Consumer.Group.Rebalance.Timeout / time.Millisecond)
+	}
+
+	meta := &ConsumerGroupMemberMetadata{
+		Topics:   topics,
+		UserData: c.config.Consumer.Group.Member.UserData,
+	}
+	strategy := c.config.Consumer.Group.Rebalance.Strategy
+	if err := req.AddGroupProtocolMetadata(strategy.Name(), meta); err != nil {
+		return nil, err
+	}
+
+	return coordinator.JoinGroup(req)
+}
+
+func (c *consumerGroup) syncGroupRequest(coordinator *Broker, plan BalanceStrategyPlan, generationID int32) (*SyncGroupResponse, error) {
+	req := &SyncGroupRequest{
+		GroupId:      c.groupID,
+		MemberId:     c.memberID,
+		GenerationId: generationID,
+	}
+	for memberID, topics := range plan {
+		err := req.AddGroupAssignmentMember(memberID, &ConsumerGroupMemberAssignment{
+			Topics: topics,
+		})
+		if err != nil {
+			return nil, err
+		}
+	}
+	return coordinator.SyncGroup(req)
+}
+
+func (c *consumerGroup) heartbeatRequest(coordinator *Broker, memberID string, generationID int32) (*HeartbeatResponse, error) {
+	req := &HeartbeatRequest{
+		GroupId:      c.groupID,
+		MemberId:     memberID,
+		GenerationId: generationID,
+	}
+
+	return coordinator.Heartbeat(req)
+}
+
+func (c *consumerGroup) balance(members map[string]ConsumerGroupMemberMetadata) (BalanceStrategyPlan, error) {
+	topics := make(map[string][]int32)
+	for _, meta := range members {
+		for _, topic := range meta.Topics {
+			topics[topic] = nil
+		}
+	}
+
+	for topic := range topics {
+		partitions, err := c.client.Partitions(topic)
+		if err != nil {
+			return nil, err
+		}
+		topics[topic] = partitions
+	}
+
+	strategy := c.config.Consumer.Group.Rebalance.Strategy
+	return strategy.Plan(members, topics)
+}
+
+// Leaves the cluster, called by Close, protected by lock.
+func (c *consumerGroup) leave() error {
+	if c.memberID == "" {
+		return nil
+	}
+
+	coordinator, err := c.client.Coordinator(c.groupID)
+	if err != nil {
+		return err
+	}
+
+	resp, err := coordinator.LeaveGroup(&LeaveGroupRequest{
+		GroupId:  c.groupID,
+		MemberId: c.memberID,
+	})
+	if err != nil {
+		_ = coordinator.Close()
+		return err
+	}
+
+	// Unset memberID
+	c.memberID = ""
+
+	// Check response
+	switch resp.Err {
+	case ErrRebalanceInProgress, ErrUnknownMemberId, ErrNoError:
+		return nil
+	default:
+		return resp.Err
+	}
+}
+
+func (c *consumerGroup) handleError(err error, topic string, partition int32) {
+	select {
+	case <-c.closed:
+		return
+	default:
+	}
+
+	if _, ok := err.(*ConsumerError); !ok && topic != "" && partition > -1 {
+		err = &ConsumerError{
+			Topic:     topic,
+			Partition: partition,
+			Err:       err,
+		}
+	}
+
+	if c.config.Consumer.Return.Errors {
+		select {
+		case c.errors <- err:
+		default:
+		}
+	} else {
+		Logger.Println(err)
+	}
+}
+
+// --------------------------------------------------------------------
+
+// ConsumerGroupSession represents a consumer group member session.
+type ConsumerGroupSession interface {
+	// Claims returns information about the claimed partitions by topic.
+	Claims() map[string][]int32
+
+	// MemberID returns the cluster member ID.
+	MemberID() string
+
+	// GenerationID returns the current generation ID.
+	GenerationID() int32
+
+	// MarkOffset marks the provided offset, alongside a metadata string
+	// that represents the state of the partition consumer at that point in time. The
+	// metadata string can be used by another consumer to restore that state, so it
+	// can resume consumption.
+	//
+	// To follow upstream conventions, you are expected to mark the offset of the
+	// next message to read, not the last message read. Thus, when calling `MarkOffset`
+	// you should typically add one to the offset of the last consumed message.
+	//
+	// Note: calling MarkOffset does not necessarily commit the offset to the backend
+	// store immediately for efficiency reasons, and it may never be committed if
+	// your application crashes. This means that you may end up processing the same
+	// message twice, and your processing should ideally be idempotent.
+	MarkOffset(topic string, partition int32, offset int64, metadata string)
+
+	// ResetOffset resets to the provided offset, alongside a metadata string that
+	// represents the state of the partition consumer at that point in time. Reset
+	// acts as a counterpart to MarkOffset, the difference being that it allows to
+	// reset an offset to an earlier or smaller value, where MarkOffset only
+	// allows incrementing the offset. cf MarkOffset for more details.
+	ResetOffset(topic string, partition int32, offset int64, metadata string)
+
+	// MarkMessage marks a message as consumed.
+	MarkMessage(msg *ConsumerMessage, metadata string)
+
+	// Context returns the session context.
+	Context() context.Context
+}
+
+type consumerGroupSession struct {
+	parent       *consumerGroup
+	memberID     string
+	generationID int32
+	handler      ConsumerGroupHandler
+
+	claims  map[string][]int32
+	offsets *offsetManager
+	ctx     context.Context
+	cancel  func()
+
+	waitGroup       sync.WaitGroup
+	releaseOnce     sync.Once
+	hbDying, hbDead chan none
+}
+
+func newConsumerGroupSession(parent *consumerGroup, ctx context.Context, claims map[string][]int32, memberID string, generationID int32, handler ConsumerGroupHandler) (*consumerGroupSession, error) {
+	// init offset manager
+	offsets, err := newOffsetManagerFromClient(parent.groupID, memberID, generationID, parent.client)
+	if err != nil {
+		return nil, err
+	}
+
+	// init context
+	ctx, cancel := context.WithCancel(ctx)
+
+	// init session
+	sess := &consumerGroupSession{
+		parent:       parent,
+		memberID:     memberID,
+		generationID: generationID,
+		handler:      handler,
+		offsets:      offsets,
+		claims:       claims,
+		ctx:          ctx,
+		cancel:       cancel,
+		hbDying:      make(chan none),
+		hbDead:       make(chan none),
+	}
+
+	// start heartbeat loop
+	go sess.heartbeatLoop()
+
+	// create a POM for each claim
+	for topic, partitions := range claims {
+		for _, partition := range partitions {
+			pom, err := offsets.ManagePartition(topic, partition)
+			if err != nil {
+				_ = sess.release(false)
+				return nil, err
+			}
+
+			// handle POM errors
+			go func(topic string, partition int32) {
+				for err := range pom.Errors() {
+					sess.parent.handleError(err, topic, partition)
+				}
+			}(topic, partition)
+		}
+	}
+
+	// perform setup
+	if err := handler.Setup(sess); err != nil {
+		_ = sess.release(true)
+		return nil, err
+	}
+
+	// start consuming
+	for topic, partitions := range claims {
+		for _, partition := range partitions {
+			sess.waitGroup.Add(1)
+
+			go func(topic string, partition int32) {
+				defer sess.waitGroup.Done()
+
+				// cancel the as session as soon as the first
+				// goroutine exits
+				defer sess.cancel()
+
+				// consume a single topic/partition, blocking
+				sess.consume(topic, partition)
+			}(topic, partition)
+		}
+	}
+	return sess, nil
+}
+
+func (s *consumerGroupSession) Claims() map[string][]int32 { return s.claims }
+func (s *consumerGroupSession) MemberID() string           { return s.memberID }
+func (s *consumerGroupSession) GenerationID() int32        { return s.generationID }
+
+func (s *consumerGroupSession) MarkOffset(topic string, partition int32, offset int64, metadata string) {
+	if pom := s.offsets.findPOM(topic, partition); pom != nil {
+		pom.MarkOffset(offset, metadata)
+	}
+}
+
+func (s *consumerGroupSession) ResetOffset(topic string, partition int32, offset int64, metadata string) {
+	if pom := s.offsets.findPOM(topic, partition); pom != nil {
+		pom.ResetOffset(offset, metadata)
+	}
+}
+
+func (s *consumerGroupSession) MarkMessage(msg *ConsumerMessage, metadata string) {
+	s.MarkOffset(msg.Topic, msg.Partition, msg.Offset+1, metadata)
+}
+
+func (s *consumerGroupSession) Context() context.Context {
+	return s.ctx
+}
+
+func (s *consumerGroupSession) consume(topic string, partition int32) {
+	// quick exit if rebalance is due
+	select {
+	case <-s.ctx.Done():
+		return
+	case <-s.parent.closed:
+		return
+	default:
+	}
+
+	// get next offset
+	offset := s.parent.config.Consumer.Offsets.Initial
+	if pom := s.offsets.findPOM(topic, partition); pom != nil {
+		offset, _ = pom.NextOffset()
+	}
+
+	// create new claim
+	claim, err := newConsumerGroupClaim(s, topic, partition, offset)
+	if err != nil {
+		s.parent.handleError(err, topic, partition)
+		return
+	}
+
+	// handle errors
+	go func() {
+		for err := range claim.Errors() {
+			s.parent.handleError(err, topic, partition)
+		}
+	}()
+
+	// trigger close when session is done
+	go func() {
+		select {
+		case <-s.ctx.Done():
+		case <-s.parent.closed:
+		}
+		claim.AsyncClose()
+	}()
+
+	// start processing
+	if err := s.handler.ConsumeClaim(s, claim); err != nil {
+		s.parent.handleError(err, topic, partition)
+	}
+
+	// ensure consumer is clased & drained
+	claim.AsyncClose()
+	for _, err := range claim.waitClosed() {
+		s.parent.handleError(err, topic, partition)
+	}
+}
+
+func (s *consumerGroupSession) release(withCleanup bool) (err error) {
+	// signal release, stop heartbeat
+	s.cancel()
+
+	// wait for consumers to exit
+	s.waitGroup.Wait()
+
+	// perform release
+	s.releaseOnce.Do(func() {
+		if withCleanup {
+			if e := s.handler.Cleanup(s); e != nil {
+				s.parent.handleError(err, "", -1)
+				err = e
+			}
+		}
+
+		if e := s.offsets.Close(); e != nil {
+			err = e
+		}
+
+		close(s.hbDying)
+		<-s.hbDead
+	})
+
+	return
+}
+
+func (s *consumerGroupSession) heartbeatLoop() {
+	defer close(s.hbDead)
+	defer s.cancel() // trigger the end of the session on exit
+
+	pause := time.NewTicker(s.parent.config.Consumer.Group.Heartbeat.Interval)
+	defer pause.Stop()
+
+	retries := s.parent.config.Metadata.Retry.Max
+	for {
+		coordinator, err := s.parent.client.Coordinator(s.parent.groupID)
+		if err != nil {
+			if retries <= 0 {
+				s.parent.handleError(err, "", -1)
+				return
+			}
+
+			select {
+			case <-s.hbDying:
+				return
+			case <-time.After(s.parent.config.Metadata.Retry.Backoff):
+				retries--
+			}
+			continue
+		}
+
+		resp, err := s.parent.heartbeatRequest(coordinator, s.memberID, s.generationID)
+		if err != nil {
+			_ = coordinator.Close()
+			retries--
+			continue
+		}
+
+		switch resp.Err {
+		case ErrNoError:
+			retries = s.parent.config.Metadata.Retry.Max
+		case ErrRebalanceInProgress, ErrUnknownMemberId, ErrIllegalGeneration:
+			return
+		default:
+			s.parent.handleError(err, "", -1)
+			return
+		}
+
+		select {
+		case <-pause.C:
+		case <-s.hbDying:
+			return
+		}
+	}
+}
+
+// --------------------------------------------------------------------
+
+// ConsumerGroupHandler instances are used to handle individual topic/partition claims.
+// It also provides hooks for your consumer group session life-cycle and allow you to
+// trigger logic before or after the consume loop(s).
+//
+// PLEASE NOTE that handlers are likely be called from several goroutines concurrently,
+// ensure that all state is safely protected against race conditions.
+type ConsumerGroupHandler interface {
+	// Setup is run at the beginning of a new session, before ConsumeClaim.
+	Setup(ConsumerGroupSession) error
+
+	// Cleanup is run at the end of a session, once all ConsumeClaim goroutines have exites
+	// but before the offsets are committed for the very last time.
+	Cleanup(ConsumerGroupSession) error
+
+	// ConsumeClaim must start a consumer loop of ConsumerGroupClaim's Messages().
+	// Once the Messages() channel is closed, the Handler must finish its processing
+	// loop and exit.
+	ConsumeClaim(ConsumerGroupSession, ConsumerGroupClaim) error
+}
+
+// ConsumerGroupClaim processes Kafka messages from a given topic and partition within a consumer group.
+type ConsumerGroupClaim interface {
+	// Topic returns the consumed topic name.
+	Topic() string
+
+	// Partition returns the consumed partition.
+	Partition() int32
+
+	// InitialOffset returns the initial offset that was used as a starting point for this claim.
+	InitialOffset() int64
+
+	// HighWaterMarkOffset returns the high water mark offset of the partition,
+	// i.e. the offset that will be used for the next message that will be produced.
+	// You can use this to determine how far behind the processing is.
+	HighWaterMarkOffset() int64
+
+	// Messages returns the read channel for the messages that are returned by
+	// the broker. The messages channel will be closed when a new rebalance cycle
+	// is due. You must finish processing and mark offsets within
+	// Config.Consumer.Group.Session.Timeout before the topic/partition is eventually
+	// re-assigned to another group member.
+	Messages() <-chan *ConsumerMessage
+}
+
+type consumerGroupClaim struct {
+	topic     string
+	partition int32
+	offset    int64
+	PartitionConsumer
+}
+
+func newConsumerGroupClaim(sess *consumerGroupSession, topic string, partition int32, offset int64) (*consumerGroupClaim, error) {
+	pcm, err := sess.parent.consumer.ConsumePartition(topic, partition, offset)
+	if err == ErrOffsetOutOfRange {
+		offset = sess.parent.config.Consumer.Offsets.Initial
+		pcm, err = sess.parent.consumer.ConsumePartition(topic, partition, offset)
+	}
+	if err != nil {
+		return nil, err
+	}
+
+	go func() {
+		for err := range pcm.Errors() {
+			sess.parent.handleError(err, topic, partition)
+		}
+	}()
+
+	return &consumerGroupClaim{
+		topic:             topic,
+		partition:         partition,
+		offset:            offset,
+		PartitionConsumer: pcm,
+	}, nil
+}
+
+func (c *consumerGroupClaim) Topic() string        { return c.topic }
+func (c *consumerGroupClaim) Partition() int32     { return c.partition }
+func (c *consumerGroupClaim) InitialOffset() int64 { return c.offset }
+
+// Drains messages and errors, ensures the claim is fully closed.
+func (c *consumerGroupClaim) waitClosed() (errs ConsumerErrors) {
+	go func() {
+		for range c.Messages() {
+		}
+	}()
+
+	for err := range c.Errors() {
+		errs = append(errs, err)
+	}
+	return
+}

vendor/github.com/googollee/go-socket.io/.gitignore

@@ -0,0 +1,4 @@
+
+.DS_Store
+.idea/
+go.sum

vendor/github.com/konsorten/go-windows-terminal-sequences/README.md

@@ -0,0 +1,40 @@
+# Windows Terminal Sequences
+
+This library allow for enabling Windows terminal color support for Go.
+
+See [Console Virtual Terminal Sequences](https://docs.microsoft.com/en-us/windows/console/console-virtual-terminal-sequences) for details.
+
+## Usage
+
+```go
+import (
+	"syscall"
+	
+	sequences "github.com/konsorten/go-windows-terminal-sequences"
+)
+
+func main() {
+	sequences.EnableVirtualTerminalProcessing(syscall.Stdout, true)
+}
+
+```
+
+## Authors
+
+The tool is sponsored by the [marvin + konsorten GmbH](http://www.konsorten.de).
+
+We thank all the authors who provided code to this library:
+
+* Felix Kollmann
+
+## License
+
+(The MIT License)
+
+Copyright (c) 2018 marvin + konsorten GmbH (open-source@konsorten.de)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the 'Software'), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

vendor/github.com/konsorten/go-windows-terminal-sequences/license

@@ -0,0 +1,9 @@
+(The MIT License)
+
+Copyright (c) 2017 marvin + konsorten GmbH (open-source@konsorten.de)
+
+Permission is hereby granted, free of charge, to any person obtaining a copy of this software and associated documentation files (the 'Software'), to deal in the Software without restriction, including without limitation the rights to use, copy, modify, merge, publish, distribute, sublicense, and/or sell copies of the Software, and to permit persons to whom the Software is furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED 'AS IS', WITHOUT WARRANTY OF ANY KIND, EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE SOFTWARE.

vendor/github.com/konsorten/go-windows-terminal-sequences/sequences.go

@@ -0,0 +1,36 @@
+// +build windows
+
+package sequences
+
+import (
+	"syscall"
+	"unsafe"
+)
+
+var (
+	kernel32Dll    *syscall.LazyDLL  = syscall.NewLazyDLL("Kernel32.dll")
+	setConsoleMode *syscall.LazyProc = kernel32Dll.NewProc("SetConsoleMode")
+)
+
+func EnableVirtualTerminalProcessing(stream syscall.Handle, enable bool) error {
+	const ENABLE_VIRTUAL_TERMINAL_PROCESSING uint32 = 0x4
+
+	var mode uint32
+	err := syscall.GetConsoleMode(syscall.Stdout, &mode)
+	if err != nil {
+		return err
+	}
+
+	if enable {
+		mode |= ENABLE_VIRTUAL_TERMINAL_PROCESSING
+	} else {
+		mode &^= ENABLE_VIRTUAL_TERMINAL_PROCESSING
+	}
+
+	ret, _, err := setConsoleMode.Call(uintptr(unsafe.Pointer(stream)), uintptr(mode))
+	if ret == 0 {
+		return err
+	}
+
+	return nil
+}

vendor/github.com/mohae/deepcopy/.gitignore

@@ -0,0 +1,26 @@
+# Compiled Object files, Static and Dynamic libs (Shared Objects)
+*.o
+*.a
+*.so
+
+# Folders
+_obj
+_test
+
+# Architecture specific extensions/prefixes
+*.[568vq]
+[568vq].out
+
+*.cgo1.go
+*.cgo2.c
+_cgo_defun.c
+_cgo_gotypes.go
+_cgo_export.*
+
+_testmain.go
+
+*.exe
+*.test
+*~
+*.out
+*.log

vendor/github.com/mohae/deepcopy/.travis.yml

@@ -0,0 +1,11 @@
+language: go
+
+go:
+  - tip
+
+matrix:
+  allow_failures:
+    - go: tip
+
+script:
+  - go test ./...

vendor/github.com/mohae/deepcopy/LICENSE

@@ -0,0 +1,21 @@
+The MIT License (MIT)
+
+Copyright (c) 2014 Joel
+
+Permission is hereby granted, free of charge, to any person obtaining a copy
+of this software and associated documentation files (the "Software"), to deal
+in the Software without restriction, including without limitation the rights
+to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+copies of the Software, and to permit persons to whom the Software is
+furnished to do so, subject to the following conditions:
+
+The above copyright notice and this permission notice shall be included in all
+copies or substantial portions of the Software.
+
+THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
+AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+SOFTWARE.

vendor/github.com/mohae/deepcopy/README.md

@@ -0,0 +1,8 @@
+deepCopy
+========
+[![GoDoc](https://godoc.org/github.com/mohae/deepcopy?status.svg)](https://godoc.org/github.com/mohae/deepcopy)[![Build Status](https://travis-ci.org/mohae/deepcopy.png)](https://travis-ci.org/mohae/deepcopy)
+
+DeepCopy makes deep copies of things: unexported field values are not copied.
+
+## Usage
+    cpy := deepcopy.Copy(orig)

vendor/github.com/mohae/deepcopy/deepcopy.go

@@ -0,0 +1,125 @@
+// deepcopy makes deep copies of things. A standard copy will copy the
+// pointers: deep copy copies the values pointed to.  Unexported field
+// values are not copied.
+//
+// Copyright (c)2014-2016, Joel Scoble (github.com/mohae), all rights reserved.
+// License: MIT, for more details check the included LICENSE file.
+package deepcopy
+
+import (
+	"reflect"
+	"time"
+)
+
+// Interface for delegating copy process to type
+type Interface interface {
+	DeepCopy() interface{}
+}
+
+// Iface is an alias to Copy; this exists for backwards compatibility reasons.
+func Iface(iface interface{}) interface{} {
+	return Copy(iface)
+}
+
+// Copy creates a deep copy of whatever is passed to it and returns the copy
+// in an interface{}.  The returned value will need to be asserted to the
+// correct type.
+func Copy(src interface{}) interface{} {
+	if src == nil {
+		return nil
+	}
+
+	// Make the interface a reflect.Value
+	original := reflect.ValueOf(src)
+
+	// Make a copy of the same type as the original.
+	cpy := reflect.New(original.Type()).Elem()
+
+	// Recursively copy the original.
+	copyRecursive(original, cpy)
+
+	// Return the copy as an interface.
+	return cpy.Interface()
+}
+
+// copyRecursive does the actual copying of the interface. It currently has
+// limited support for what it can handle. Add as needed.
+func copyRecursive(original, cpy reflect.Value) {
+	// check for implement deepcopy.Interface
+	if original.CanInterface() {
+		if copier, ok := original.Interface().(Interface); ok {
+			cpy.Set(reflect.ValueOf(copier.DeepCopy()))
+			return
+		}
+	}
+
+	// handle according to original's Kind
+	switch original.Kind() {
+	case reflect.Ptr:
+		// Get the actual value being pointed to.
+		originalValue := original.Elem()
+
+		// if  it isn't valid, return.
+		if !originalValue.IsValid() {
+			return
+		}
+		cpy.Set(reflect.New(originalValue.Type()))
+		copyRecursive(originalValue, cpy.Elem())
+
+	case reflect.Interface:
+		// If this is a nil, don't do anything
+		if original.IsNil() {
+			return
+		}
+		// Get the value for the interface, not the pointer.
+		originalValue := original.Elem()
+
+		// Get the value by calling Elem().
+		copyValue := reflect.New(originalValue.Type()).Elem()
+		copyRecursive(originalValue, copyValue)
+		cpy.Set(copyValue)
+
+	case reflect.Struct:
+		t, ok := original.Interface().(time.Time)
+		if ok {
+			cpy.Set(reflect.ValueOf(t))
+			return
+		}
+		// Go through each field of the struct and copy it.
+		for i := 0; i < original.NumField(); i++ {
+			// The Type's StructField for a given field is checked to see if StructField.PkgPath
+			// is set to determine if the field is exported or not because CanSet() returns false
+			// for settable fields.  I'm not sure why.  -mohae
+			if original.Type().Field(i).PkgPath != "" {
+				continue
+			}
+			copyRecursive(original.Field(i), cpy.Field(i))
+		}
+
+	case reflect.Slice:
+		if original.IsNil() {
+			return
+		}
+		// Make a new slice and copy each element.
+		cpy.Set(reflect.MakeSlice(original.Type(), original.Len(), original.Cap()))
+		for i := 0; i < original.Len(); i++ {
+			copyRecursive(original.Index(i), cpy.Index(i))
+		}
+
+	case reflect.Map:
+		if original.IsNil() {
+			return
+		}
+		cpy.Set(reflect.MakeMap(original.Type()))
+		for _, key := range original.MapKeys() {
+			originalValue := original.MapIndex(key)
+			copyValue := reflect.New(originalValue.Type()).Elem()
+			copyRecursive(originalValue, copyValue)
+			copyKey := Copy(key.Interface())
+			cpy.SetMapIndex(reflect.ValueOf(copyKey), copyValue)
+		}
+
+	default:
+		cpy.Set(original)
+	}
+}

vendor/github.com/sirupsen/logrus/go.mod

@@ -0,0 +1,10 @@
+module github.com/sirupsen/logrus
+
+require (
+	github.com/davecgh/go-spew v1.1.1 // indirect
+	github.com/konsorten/go-windows-terminal-sequences v0.0.0-20180402223658-b729f2633dfe
+	github.com/pmezard/go-difflib v1.0.0 // indirect
+	github.com/stretchr/testify v1.2.2
+	golang.org/x/crypto v0.0.0-20180904163835-0709b304e793
+	golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33
+)

vendor/github.com/sirupsen/logrus/go.sum

@@ -0,0 +1,12 @@
+github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
+github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/konsorten/go-windows-terminal-sequences v0.0.0-20180402223658-b729f2633dfe h1:CHRGQ8V7OlCYtwaKPJi3iA7J+YdNKdo8j7nG5IgDhjs=
+github.com/konsorten/go-windows-terminal-sequences v0.0.0-20180402223658-b729f2633dfe/go.mod h1:T0+1ngSBFLxvqU3pZ+m/2kptfBszLMUkC4ZK/EgS/cQ=
+github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
+github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
+github.com/stretchr/testify v1.2.2 h1:bSDNvY7ZPG5RlJ8otE/7V6gMiyenm9RtJ7IUVIAoJ1w=
+github.com/stretchr/testify v1.2.2/go.mod h1:a8OnRcib4nhh0OaRAV+Yts87kKdq0PP7pXfy6kDkUVs=
+golang.org/x/crypto v0.0.0-20180904163835-0709b304e793 h1:u+LnwYTOOW7Ukr/fppxEb1Nwz0AtPflrblfvUudpo+I=
+golang.org/x/crypto v0.0.0-20180904163835-0709b304e793/go.mod h1:6SG95UA2DQfeDnfUPMdvaQW0Q7yPrPDi9nlGo2tz2b4=
+golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33 h1:I6FyU15t786LL7oL/hn43zqTuEGr4PN7F4XJ1p4E3Y8=
+golang.org/x/sys v0.0.0-20180905080454-ebe1bf3edb33/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=

vendor/github.com/sirupsen/logrus/terminal_appengine.go

@@ -0,0 +1,13 @@
+// Based on ssh/terminal:
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+// +build appengine
+
+package logrus
+
+import "io"
+
+func initTerminal(w io.Writer) {
+}

vendor/github.com/sirupsen/logrus/terminal_check_js.go

@@ -0,0 +1,11 @@
+// +build js
+
+package logrus
+
+import (
+	"io"
+)
+
+func checkIfTerminal(w io.Writer) bool {
+	return false
+}

vendor/github.com/sirupsen/logrus/terminal_check_windows.go

@@ -0,0 +1,20 @@
+// +build !appengine,!js,windows
+
+package logrus
+
+import (
+	"io"
+	"os"
+	"syscall"
+)
+
+func checkIfTerminal(w io.Writer) bool {
+	switch v := w.(type) {
+	case *os.File:
+		var mode uint32
+		err := syscall.GetConsoleMode(syscall.Handle(v.Fd()), &mode)
+		return err == nil
+	default:
+		return false
+	}
+}

vendor/github.com/sirupsen/logrus/terminal_windows.go

@@ -0,0 +1,18 @@
+// +build !appengine,!js,windows
+
+package logrus
+
+import (
+	"io"
+	"os"
+	"syscall"
+
+	sequences "github.com/konsorten/go-windows-terminal-sequences"
+)
+
+func initTerminal(w io.Writer) {
+	switch v := w.(type) {
+	case *os.File:
+		sequences.EnableVirtualTerminalProcessing(syscall.Handle(v.Fd()), true)
+	}
+}

vendor/golang.org/x/sys/windows/asm_windows_arm.s

@@ -0,0 +1,11 @@
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+#include "textflag.h"
+
+TEXT ·getprocaddress(SB),NOSPLIT,$0
+	B	syscall·getprocaddress(SB)
+
+TEXT ·loadlibrary(SB),NOSPLIT,$0
+	B	syscall·loadlibrary(SB)

vendor/golang.org/x/sys/windows/types_windows_arm.go

@@ -0,0 +1,22 @@
+// Copyright 2018 The Go Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style
+// license that can be found in the LICENSE file.
+
+package windows
+
+type WSAData struct {
+	Version      uint16
+	HighVersion  uint16
+	Description  [WSADESCRIPTION_LEN + 1]byte
+	SystemStatus [WSASYS_STATUS_LEN + 1]byte
+	MaxSockets   uint16
+	MaxUdpDg     uint16
+	VendorInfo   *byte
+}
+
+type Servent struct {
+	Name    *byte
+	Aliases **byte
+	Port    uint16
+	Proto   *byte
+}

vendor/yunion.io/x/sqlchemy/field_update.go

@@ -54,6 +54,10 @@ func (ts *STableSpec) updateFields(dt interface{}, fields map[string]interface{}
 	cv := make(map[string]interface{}, 0)
 	dataType := dataValue.Type()
 	ts.GetUpdateColumnValue(dataType, dataValue, cv, fields)
+	if len(cv) == 0 {
+		log.Infof("Nothing update")
+		return nil
+	}
 
 	fullFields := reflectutils.FetchStructFieldNameValueInterfaces(dataValue)
 	versionFields := make([]string, 0)
@@ -62,7 +66,10 @@ func (ts *STableSpec) updateFields(dt interface{}, fields map[string]interface{}
 	indexCols := make(map[string]interface{}, 0)
 	for _, col := range ts.Columns() {
 		name := col.Name()
-		colValue := fullFields[name]
+		colValue, ok := fullFields[name]
+		if !ok {
+			continue
+		}
 		if col.IsPrimary() && !col.IsZero(colValue) {
 			primaryCols[name] = colValue
 			continue