VOL-1967 move api-server to separate repository
Change-Id: I21b85be74205805be15f8a85e53a903d16785671
diff --git a/vendor/github.com/Shopify/sarama/config.go b/vendor/github.com/Shopify/sarama/config.go
index 9495b7f..e2e6513 100644
--- a/vendor/github.com/Shopify/sarama/config.go
+++ b/vendor/github.com/Shopify/sarama/config.go
@@ -10,6 +10,7 @@
"time"
"github.com/rcrowley/go-metrics"
+ "golang.org/x/net/proxy"
)
const defaultClientID = "sarama"
@@ -57,18 +58,28 @@
// SASLMechanism is the name of the enabled SASL mechanism.
// Possible values: OAUTHBEARER, PLAIN (defaults to PLAIN).
Mechanism SASLMechanism
+ // Version is the SASL Protocol Version to use
+ // Kafka > 1.x should use V1, except on Azure EventHub which use V0
+ Version int16
// Whether or not to send the Kafka SASL handshake first if enabled
// (defaults to true). You should only set this to false if you're using
// a non-Kafka SASL proxy.
Handshake bool
- //username and password for SASL/PLAIN authentication
+ //username and password for SASL/PLAIN or SASL/SCRAM authentication
User string
Password string
+ // authz id used for SASL/SCRAM authentication
+ SCRAMAuthzID string
+ // SCRAMClientGeneratorFunc is a generator of a user provided implementation of a SCRAM
+ // client used to perform the SCRAM exchange with the server.
+ SCRAMClientGeneratorFunc func() SCRAMClient
// TokenProvider is a user-defined callback for generating
// access tokens for SASL/OAUTHBEARER auth. See the
// AccessTokenProvider interface docs for proper implementation
// guidelines.
TokenProvider AccessTokenProvider
+
+ GSSAPI GSSAPIConfig
}
// KeepAlive specifies the keep-alive period for an active network connection.
@@ -80,6 +91,14 @@
// network being dialed.
// If nil, a local address is automatically chosen.
LocalAddr net.Addr
+
+ Proxy struct {
+ // Whether or not to use proxy when connecting to the broker
+ // (defaults to false).
+ Enable bool
+ // The proxy dialer to use enabled (defaults to nil).
+ Dialer proxy.Dialer
+ }
}
// Metadata is the namespace for metadata management properties used by the
@@ -107,6 +126,13 @@
// and usually more convenient, but can take up a substantial amount of
// memory if you have many topics and partitions. Defaults to true.
Full bool
+
+ // How long to wait for a successful metadata response.
+ // Disabled by default which means a metadata request against an unreachable
+ // cluster (all brokers are unreachable or unresponsive) can take up to
+ // `Net.[Dial|Read]Timeout * BrokerCount * (Metadata.Retry.Max + 1) + Metadata.Retry.Backoff * Metadata.Retry.Max`
+ // to fail.
+ Timeout time.Duration
}
// Producer is the namespace for configuration related to producing messages,
@@ -333,6 +359,11 @@
Max int
}
}
+
+ // IsolationLevel support 2 mode:
+ // - use `ReadUncommitted` (default) to consume and return all messages in message channel
+ // - use `ReadCommitted` to hide messages that are part of an aborted transaction
+ IsolationLevel IsolationLevel
}
// A user-provided string sent with every request to the brokers for logging,
@@ -370,6 +401,7 @@
c.Net.ReadTimeout = 30 * time.Second
c.Net.WriteTimeout = 30 * time.Second
c.Net.SASL.Handshake = true
+ c.Net.SASL.Version = SASLHandshakeV0
c.Metadata.Retry.Max = 3
c.Metadata.Retry.Backoff = 250 * time.Millisecond
@@ -414,10 +446,10 @@
// ConfigurationError if the specified values don't make sense.
func (c *Config) Validate() error {
// some configuration values should be warned on but not fail completely, do those first
- if c.Net.TLS.Enable == false && c.Net.TLS.Config != nil {
+ if !c.Net.TLS.Enable && c.Net.TLS.Config != nil {
Logger.Println("Net.TLS is disabled but a non-nil configuration was provided.")
}
- if c.Net.SASL.Enable == false {
+ if !c.Net.SASL.Enable {
if c.Net.SASL.User != "" {
Logger.Println("Net.SASL is disabled but a non-empty username was provided.")
}
@@ -475,22 +507,62 @@
case c.Net.KeepAlive < 0:
return ConfigurationError("Net.KeepAlive must be >= 0")
case c.Net.SASL.Enable:
- // For backwards compatibility, empty mechanism value defaults to PLAIN
- isSASLPlain := len(c.Net.SASL.Mechanism) == 0 || c.Net.SASL.Mechanism == SASLTypePlaintext
- if isSASLPlain {
+ if c.Net.SASL.Mechanism == "" {
+ c.Net.SASL.Mechanism = SASLTypePlaintext
+ }
+
+ switch c.Net.SASL.Mechanism {
+ case SASLTypePlaintext:
if c.Net.SASL.User == "" {
return ConfigurationError("Net.SASL.User must not be empty when SASL is enabled")
}
if c.Net.SASL.Password == "" {
return ConfigurationError("Net.SASL.Password must not be empty when SASL is enabled")
}
- } else if c.Net.SASL.Mechanism == SASLTypeOAuth {
+ case SASLTypeOAuth:
if c.Net.SASL.TokenProvider == nil {
- return ConfigurationError("An AccessTokenProvider instance must be provided to Net.SASL.User.TokenProvider")
+ return ConfigurationError("An AccessTokenProvider instance must be provided to Net.SASL.TokenProvider")
}
- } else {
- msg := fmt.Sprintf("The SASL mechanism configuration is invalid. Possible values are `%s` and `%s`",
- SASLTypeOAuth, SASLTypePlaintext)
+ case SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512:
+ if c.Net.SASL.User == "" {
+ return ConfigurationError("Net.SASL.User must not be empty when SASL is enabled")
+ }
+ if c.Net.SASL.Password == "" {
+ return ConfigurationError("Net.SASL.Password must not be empty when SASL is enabled")
+ }
+ if c.Net.SASL.SCRAMClientGeneratorFunc == nil {
+ return ConfigurationError("A SCRAMClientGeneratorFunc function must be provided to Net.SASL.SCRAMClientGeneratorFunc")
+ }
+ case SASLTypeGSSAPI:
+ if c.Net.SASL.GSSAPI.ServiceName == "" {
+ return ConfigurationError("Net.SASL.GSSAPI.ServiceName must not be empty when GSS-API mechanism is used")
+ }
+
+ if c.Net.SASL.GSSAPI.AuthType == KRB5_USER_AUTH {
+ if c.Net.SASL.GSSAPI.Password == "" {
+ return ConfigurationError("Net.SASL.GSSAPI.Password must not be empty when GSS-API " +
+ "mechanism is used and Net.SASL.GSSAPI.AuthType = KRB5_USER_AUTH")
+ }
+ } else if c.Net.SASL.GSSAPI.AuthType == KRB5_KEYTAB_AUTH {
+ if c.Net.SASL.GSSAPI.KeyTabPath == "" {
+ return ConfigurationError("Net.SASL.GSSAPI.KeyTabPath must not be empty when GSS-API mechanism is used" +
+ " and Net.SASL.GSSAPI.AuthType = KRB5_KEYTAB_AUTH")
+ }
+ } else {
+ return ConfigurationError("Net.SASL.GSSAPI.AuthType is invalid. Possible values are KRB5_USER_AUTH and KRB5_KEYTAB_AUTH")
+ }
+ if c.Net.SASL.GSSAPI.KerberosConfigPath == "" {
+ return ConfigurationError("Net.SASL.GSSAPI.KerberosConfigPath must not be empty when GSS-API mechanism is used")
+ }
+ if c.Net.SASL.GSSAPI.Username == "" {
+ return ConfigurationError("Net.SASL.GSSAPI.Username must not be empty when GSS-API mechanism is used")
+ }
+ if c.Net.SASL.GSSAPI.Realm == "" {
+ return ConfigurationError("Net.SASL.GSSAPI.Realm must not be empty when GSS-API mechanism is used")
+ }
+ default:
+ msg := fmt.Sprintf("The SASL mechanism configuration is invalid. Possible values are `%s`, `%s`, `%s`, `%s` and `%s`",
+ SASLTypeOAuth, SASLTypePlaintext, SASLTypeSCRAMSHA256, SASLTypeSCRAMSHA512, SASLTypeGSSAPI)
return ConfigurationError(msg)
}
}
@@ -584,6 +656,13 @@
return ConfigurationError("Consumer.Offsets.Initial must be OffsetOldest or OffsetNewest")
case c.Consumer.Offsets.Retry.Max < 0:
return ConfigurationError("Consumer.Offsets.Retry.Max must be >= 0")
+ case c.Consumer.IsolationLevel != ReadUncommitted && c.Consumer.IsolationLevel != ReadCommitted:
+ return ConfigurationError("Consumer.IsolationLevel must be ReadUncommitted or ReadCommitted")
+ }
+
+ // validate IsolationLevel
+ if c.Consumer.IsolationLevel == ReadCommitted && !c.Version.IsAtLeast(V0_11_0_0) {
+ return ConfigurationError("ReadCommitted requires Version >= V0_11_0_0")
}
// validate the Consumer Group values