Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Adding TCP keepalives support for the broker's connection #408

Merged
merged 6 commits into from
Apr 6, 2015
Merged

Adding TCP keepalives support for the broker's connection #408

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
76aa551
adding tcp keepalives to the broker's socket connection
Apr 3, 2015
3a63002
removing the wrapper function for creating tcp connection
Apr 3, 2015
a816b42
addressing comments made during PR
Apr 6, 2015
4fb06d6
removing the keepaive setting from client_test
Apr 6, 2015
8c2e14c
adding validation for the tpc keepalive client config
Apr 6, 2015
449c713
Adding equals sign to the invalid config message
epsniff Apr 6, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions CHANGELOG.md
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -7,6 +7,7 @@ Improvements:
([#389](https://github.com/Shopify/sarama/pull/389)).
- The consumer produces much more useful logging output when leadership
changes ([#385](https://github.com/Shopify/sarama/pull/385)).
- Added support for tcp keepalives ([#407](https://github.com/Shopify/sarama/issues/407)).

Bug Fixes:
- The OffsetCommitRequest message now correctly implements all three possible
Expand Down
7 changes: 6 additions & 1 deletion broker.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -68,7 +68,12 @@ func (b *Broker) Open(conf *Config) error {
go withRecover(func() {
defer b.lock.Unlock()

b.conn, b.connErr = net.DialTimeout("tcp", b.addr, conf.Net.DialTimeout)
dialer := net.Dialer{
Timeout: conf.Net.DialTimeout,
KeepAlive: conf.Net.KeepAlive,
}

b.conn, b.connErr = dialer.Dial("tcp", b.addr)
if b.connErr != nil {
b.conn = nil
atomic.StoreInt32(&b.opened, 0)
Expand Down
6 changes: 6 additions & 0 deletions config.go
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,10 @@ type Config struct {
DialTimeout time.Duration // How long to wait for the initial connection to succeed before timing out and returning an error (default 30s).
ReadTimeout time.Duration // How long to wait for a response before timing out and returning an error (default 30s).
WriteTimeout time.Duration // How long to wait for a transmit to succeed before timing out and returning an error (default 30s).

// KeepAlive specifies the keep-alive period for an active network connection.
// If zero, keep-alives are disabled. (default is 0: disabled).
KeepAlive time.Duration
}

// Metadata is the namespace for metadata management properties used by the Client, and shared by the Producer/Consumer.
Expand Down Expand Up @@ -186,6 +190,8 @@ func (c *Config) Validate() error {
return ConfigurationError("Invalid Net.ReadTimeout, must be > 0")
case c.Net.WriteTimeout <= 0:
return ConfigurationError("Invalid Net.WriteTimeout, must be > 0")
case c.Net.KeepAlive < 0:
return ConfigurationError("Invalid Net.KeepAlive, must be >= 0")
}

// validate the Metadata values
Expand Down