Renamed ReceiveDeadline -> HeartbeatTimeout

Author: ekoutanov

config.go

@@ -36,7 +36,7 @@ type Config struct {
 	Name                  string
 	PollDuration          *time.Duration
 	MinPollInterval       *time.Duration
-	ReceiveDeadline       *time.Duration
+	HeartbeatTimeout      *time.Duration
 }
 
 // Validate the Config, returning an error if invalid.
@@ -52,7 +52,7 @@ func (c Config) Validate() error {
 		validation.Field(&c.Name, validation.Required, validation.Match(regexp.MustCompile("^[^%]*$"))),
 		validation.Field(&c.PollDuration, validation.Required, validation.Min(1*time.Millisecond)),
 		validation.Field(&c.MinPollInterval, validation.Required, validation.Min(1*time.Millisecond)),
-		validation.Field(&c.ReceiveDeadline, validation.Required, validation.Min(1*time.Millisecond)),
+		validation.Field(&c.HeartbeatTimeout, validation.Required, validation.Min(1*time.Millisecond)),
 	)
 }
 
@@ -68,7 +68,7 @@ func (c Config) String() string {
 		", Name=", c.Name,
 		", PollDuration=", c.PollDuration,
 		", MinPollInterval=", c.MinPollInterval,
-		", ReceiveDeadline=", c.ReceiveDeadline, "]")
+		", HeartbeatTimeout=", c.HeartbeatTimeout, "]")
 }
 
 const (
@@ -78,8 +78,8 @@ const (
 	// DefaultMinPollInterval is the default value of Config.MinPollInterval
 	DefaultMinPollInterval = 100 * time.Millisecond
 
-	// DefaultReceiveDeadline is the default value of Config.ReceiveDeadline
-	DefaultReceiveDeadline = 5 * time.Second
+	// DefaultHeartbeatTimeout is the default value of Config.HeartbeatTimeout
+	DefaultHeartbeatTimeout = 5 * time.Second
 )
 
 // SetDefaults assigns the default values to optional fields.
@@ -111,7 +111,7 @@ func (c *Config) SetDefaults() {
 
 	defaultDuration(&c.PollDuration, DefaultPollDuration)
 	defaultDuration(&c.MinPollInterval, DefaultMinPollInterval)
-	defaultDuration(&c.ReceiveDeadline, DefaultReceiveDeadline)
+	defaultDuration(&c.HeartbeatTimeout, DefaultHeartbeatTimeout)
 }
 
 const validKafkaNameChars = "a-zA-Z0-9\\._\\-"

neli.go

@@ -89,15 +89,15 @@ func New(config Config, barrier ...Barrier) (Neli, error) {
 	err := setKafkaConfigs(consumerConfigs, KafkaConfigMap{
 		"group.id":           n.config.LeaderGroupID,
 		"enable.auto.commit": false,
-		"session.timeout.ms": int(config.ReceiveDeadline.Milliseconds()) * 3,
+		"session.timeout.ms": int(config.HeartbeatTimeout.Milliseconds()) * 3,
 	})
 	if err != nil {
 		return nil, err
 	}
 
 	producerConfigs := copyKafkaConfig(n.config.KafkaConfig)
 	err = setKafkaConfigs(producerConfigs, KafkaConfigMap{
-		"delivery.timeout.ms": int(config.ReceiveDeadline.Milliseconds()),
+		"delivery.timeout.ms": int(config.HeartbeatTimeout.Milliseconds()),
 		"linger.ms":           0,
 	})
 	if err != nil {
@@ -288,7 +288,7 @@ func (n *neli) tryPulse() (bool, error) {
 					// If we were previously the leeder, need to make sure that we are still receiving heartbeats.
 					// This enables us to detect network partitions and broker failures.
 					lastReceived := time.Unix(0, n.lastReceived.Get())
-					if elapsed := time.Now().Sub(lastReceived); elapsed > *n.config.ReceiveDeadline {
+					if elapsed := time.Now().Sub(lastReceived); elapsed > *n.config.HeartbeatTimeout {
 						n.logger().I()("Fenced leader (heartbeat timed out)")
 						n.isLeader.Set(0)
 						n.barrier(LeaderFenced{})

neli_test.go

@@ -35,7 +35,7 @@ func (fixtureOpts fixtures) create() (scribe.MockScribe, *consMock, *prodMock, C
 		LeaderTopic:           "test.topic",
 		MinPollInterval:       Duration(1 * time.Millisecond),
 		PollDuration:          Duration(1 * time.Millisecond),
-		ReceiveDeadline:       Duration(10 * time.Second),
+		HeartbeatTimeout:      Duration(10 * time.Second),
 	}
 	config.Scribe.SetEnabled(scribe.All)
 
@@ -328,7 +328,7 @@ func TestLeaderElectionAndRevocation_nopBarrier(t *testing.T) {
 
 func TestLeaderElectionAndRevocation_timeoutAndReconnect(t *testing.T) {
 	m, cons, _, config, b := fixtures{}.create()
-	config.ReceiveDeadline = Duration(1 * time.Millisecond)
+	config.HeartbeatTimeout = Duration(1 * time.Millisecond)
 
 	n, err := New(config, b.barrier())
 	require.Nil(t, err)