We'd like to be able to override how a consumer queries from NSQLookupds. I think this approach would also allow others to switch NSQLookupd for etcd, zookeeper, etc.

This is still a WIP but I hope you can see where I'm going with this. Suggestions and Criticism welcome.

cc @oliver-bitly @jehiah @mreiferson

I'm guessing this is user error, but I'm getting a panic when stopping a consumer as of this commit: 4e74fa1f8933064a4f04786e65d3ee7b611be598. Reading the comment in there, I'm trying to understand what's going on.

My shutdown code looks something like this:

func (n *NSQPeer) Teardown() {
    n.producer.Stop()
    if n.consumer != nil {
        n.consumer.DisconnectFromNSQD(n.host)
        n.consumer.Stop()
        <-n.consumer.StopChan
    }
}

Here's the error:

runtime.panic(0x509600, 0x84c1d5)
        /usr/local/Cellar/go/1.3/libexec/src/pkg/runtime/panic.c:279 +0xf5
github.com/bitly/go-nsq.(*Consumer).exit(0xc20800ef00)
        /Users/tylertreat/Go/src/github.com/bitly/go-nsq/consumer.go:1082 +0x2e
github.com/bitly/go-nsq.func·007()
        /Users/tylertreat/Go/src/github.com/bitly/go-nsq/consumer.go:990 +0x29
created by time.goFunc
        /usr/local/Cellar/go/1.3/libexec/src/pkg/time/sleep.go:121 +0x47

Can anyone provide some insight on why I'm seeing this now?

1. env

nsqd --version nsqd v1.0.0-compat (built w/go1.8)

go-nsq version commit b9762cdcb6d5cc5ac5287ca076354143d332cc97 Tue Feb 14 16:13:23 2017 -0800

# go version go version go1.8 linux/amd64

2. what do i meet

• Program runed, but channel not created at http://127.0.0.1:4151/stats
• Sometimes this will appear, it always appear at program starts. And when appear, program will never( wait for 2days) got any message.
• It always appear at program starts. When program successfully starts, seems won't have a chance to reproduct.
• I have serveral topic to listen. when it appears in one topic, other topic may not affected.
• When set go-nsq debug on, program will print [d.c2.cp#ephemeral/detect#ephemeral] (127.0.0.3:4150) heartbeat received. but seems won't got any other nsq message.

# curl -XGET http://127.0.0.1:4151/stats; 
[[email protected] ptd]# curl -XGET http://127.0.0.1:4151/stats; 
nsqd v1.0.0-compat (built w/go1.8)
start_time 2017-03-27T19:07:02+08:00
uptime 45m7.260003591s

Health: OK

   [.raw.raw#ephemeral] depth: 0     be-depth: 0     msgs: 6656     e2e%: 
      [processor#ephemeral      ] depth: 0     be-depth: 0     inflt: 0    def: 0    re-q: 0     timeout: 0     msgs: 6656     e2e%: 
        [V2 18-190               ] state: 3 inflt: 0    rdy: 1    fin: 6656     re-q: 0        msgs: 6656     connected: 8s

   [d.c2.cp#ephemeral] depth: 10000 be-depth: 0     msgs: 791227   e2e%: 

   [d.c2.cp.done#ephemeral] depth: 0     be-depth: 0     msgs: 0        e2e%: 
      [processor#ephemeral      ] depth: 0     be-depth: 0     inflt: 0    def: 0    re-q: 0     timeout: 0     msgs: 0        e2e%: 
        [V2 18-190               ] state: 3 inflt: 0    rdy: 1    fin: 0        re-q: 0        msgs: 0        connected: 8s

   [d.c2.domain#ephemeral] depth: 10000 be-depth: 0     msgs: 208218   e2e%: 

...

code snip

func nsqSubscribe(addr string, topic string, channel string, hdlr nsq.HandlerFunc) error {
    consumer, err := nsq.NewConsumer(topic, channel, nsq.NewConfig())
    if err != nil {
        print("new consumer error: ", err, "\n")
        time.Sleep(1 * time.Second) //wait 1s
        panic(err)
    }   
    consumer.AddHandler(hdlr)
    err = consumer.ConnectToNSQD(addr)
    if err != nil {
        print("connect nsqd error: ", err, "\n")
        time.Sleep(1 * time.Second) //wait 1s
        panic(err)
    }   
    _ = <-consumer.StopChan
    panic("nsq conn dead topic=" + topic + " channel=" + channel)
    return nil 
}

func main(){
producer, err := nsq.NewProducer(nsqConf.Local.Addr, nsq.NewConfig())
    if err != nil {
        panic(err)
    }
    go func() {
        detector := new(c2.C2Sdk)

        if detectorEnabled {
            err = detector.Init()
            if err != nil {
                fmt.Println("Failed to init c2")
                panic(err)
            }
        }
        nsqSubscribe(nsqConf.Local.Addr, "d.c2.cp#ephemeral", "detect#ephemeral",
            nsq.HandlerFunc(func(message *nsq.Message) error {
                return handler_scan(message, detector,
                    producer, unmarshal_url, scan_url,
                    "d.c2.cp.done#ephemeral")
            }))
    }()
go func() {
        detector := new(c2.C2Sdk)

        if detectorEnabled {
            err = detector.Init()
            if err != nil {
                fmt.Println("Failed to init c2")
                panic(err)
            }
        } 
        nsqSubscribe(nsqConf.Local.Addr, "d.c2.url#ephemeral", "detect#ephemeral",
            nsq.HandlerFunc(func(message *nsq.Message) error {
                return handler_scan(message, detector,
                    producer, unmarshal_url, scan_url,
                    "d.c2.url.done#ephemeral")
            }))
    }()

.....

}

3. how to reproduct

• set program connect to a busy nsq server
• start & restart it for serveral times. ( 5-10 times). when not repoduct, retry.

I'm currently investigating an issue where even after calling either DisconnectFromNSQLookupd or DisconnectFromNSQD on a host that is already disconnected, re-connection attempts still occur.

In our scenario we are consuming from 3 hosts: 10.0.0.1/10.0.0.2/10.0.0.3 and the host 10.0.0.1 becomes unhealthy (service stopped) we remove it from the list of nodes and call DisconnectFromNSQLookupd or DisconnectFromNSQD.

As you can see from the following log output, the connection to host 10.0.0.1 was lost and re-connection attempts are made. The host is then removed from our list and DisconnectFromNSQLookupd is called. However, the re-connection attempts still occur:

2016-04-29 14:35:26 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:35:26 [DBG] IDENTIFY response: &{MaxRdyCount:2500 TLSv1:false Deflate:false Snappy:false AuthRequired:false} (conn.go 328)
2016-04-29 14:35:29 [ERR] IO error - EOF (conn.go 471)
2016-04-29 14:35:29 [INF] beginning close (conn.go 611)
2016-04-29 14:35:29 [INF] readLoop exiting (conn.go 528)
2016-04-29 14:35:29 [INF] breaking out of writeLoop (conn.go 535)
2016-04-29 14:35:29 [INF] writeLoop exiting (conn.go 581)
2016-04-29 14:35:29 [INF] finished draining, cleanup exiting (conn.go 660)
2016-04-29 14:35:29 [INF] clean close complete (conn.go 668)
2016-04-29 14:35:29 [DBG] there are 2 connections left alive (consumer.go 741)
2016-04-29 14:35:29 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:35:44 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:35:44 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:35:44 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:35:59 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:35:59 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:35:59 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:36:14 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:36:14 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:36:14 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:36:26 [DBG] querying nsqlookupd http://10.0.0.3:4161/lookup?topic=helloworld%23ephemeral (consumer.go 419)
2016-04-29 14:36:29 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:36:29 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:36:29 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:36:44 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:36:44 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:36:44 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:36:59 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:36:59 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:36:59 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:37:14 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:37:14 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:37:14 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:37:26 [DBG] querying nsqlookupd http://10.0.0.2:4161/lookup?topic=helloworld%23ephemeral (consumer.go 419)
2016-04-29 14:37:29 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:37:29 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:37:29 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:37:32 [INF] Connecting to NSQ via lookupd hosts: [10.0.0.3:4161 10.0.0.2:4161] (subscriber.go 166)
2016-04-29 14:37:32 [INF] Connecting to NSQ via lookupd hosts: [10.0.0.3:4161 10.0.0.2:4161] (subscriber.go 166)
2016-04-29 14:37:32 [INF] Disconnecting from NSQ hosts: [10.0.0.1:4150] (subscriber.go 176)
2016-04-29 14:37:32 [INF] Disconnecting from NSQ lookupds: [10.0.0.1:4161] (subscriber.go 187)
2016-04-29 14:37:44 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:37:44 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:37:44 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:37:59 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:37:59 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:37:59 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:38:14 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:38:14 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:38:14 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)
2016-04-29 14:38:26 [DBG] querying nsqlookupd http://10.0.0.3:4161/lookup?topic=helloworld%23ephemeral (consumer.go 419)
2016-04-29 14:38:29 [INF] (10.0.0.1:4150) connecting to nsqd (consumer.go 508)
2016-04-29 14:38:29 [ERR] (10.0.0.1:4150) error connecting to nsqd - dial tcp 10.0.0.1:4150: getsockopt: connection refused (consumer.go 790)
2016-04-29 14:38:29 [DBG] (10.0.0.1:4150) re-connecting in 15 seconds... (consumer.go 776)

Is this intended/expected behaviour?

If a consumer is in backoff and attempts to recover with RDY 1 and then a message, but response with RequeueWithoutBackoff it will stall and not exit backoff. (failing test attached)

cc: @georgicodes

I did some benchmark works for nsq, and encounter some strange problem. I pub 100000 msgs to nsq by go-nsq, but always failed. and I check small count (like 10000 msgs) was ok. the go-nsq log show write to nsqd timeout, and nsqd log tell me the producer's connection be reset. what problem in such situation?

go-nsq log:

2019/02/26 22:44:52 INF    1 (192.168.1.125:4150) connecting to nsqd

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 INF    1 (192.168.1.125:4150) beginning close

2019/02/26 22:45:01 INF    1 (192.168.1.125:4150) readLoop exiting

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) IO error - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 ERR    1 (192.168.1.125:4150) sending command - write tcp 192.168.1.99:4908->192.168.1.125:4150: i/o timeout

2019/02/26 22:45:01 INF    1 (192.168.1.125:4150) breaking out of writeLoop

2019/02/26 22:45:01 INF    1 (192.168.1.125:4150) writeLoop exiting

2019/02/26 22:45:01 INF    1 (192.168.1.125:4150) finished draining, cleanup exiting

2019/02/26 22:45:01 INF    1 (192.168.1.125:4150) clean close complete

2019/02/26 22:45:01 INF    1 exiting router

nsqd log:

[nsqd] 2019/02/26 22:39:55.590775 INFO: TCP: new client(192.168.1.99:4908)
[nsqd] 2019/02/26 22:39:55.590882 INFO: CLIENT(192.168.1.99:4908): desired protocol magic '  V2'
[nsqd] 2019/02/26 22:39:55.591299 INFO: [192.168.1.99:4908] IDENTIFY: {ClientID:HuangWei-PC Hostname:HuangWei-PC HeartbeatInterval:30000 OutputBufferSize:16384 OutputBufferTimeout:250 FeatureNegotiation:true TLSv1:false Deflate:false DeflateLevel:6 Snappy:false SampleRate:0 UserAgent:go-nsq/1.0.7 MsgTimeout:0}
[nsqd] 2019/02/26 22:40:04.104850 INFO: PROTOCOL(V2): [192.168.1.99:4908] exiting ioloop
[nsqd] 2019/02/26 22:40:04.104955 ERROR: client(192.168.1.99:4908) - failed to send response - write tcp 192.168.1.125:4150->192.168.1.99:4908: write: connection reset by peer
[nsqd] 2019/02/26 22:40:04.104980 INFO: PROTOCOL(V2): [192.168.1.99:4908] exiting messagePump

go benchmark code:

func BenchmarkProducer(b *testing.B) {
	b.StopTimer()

	config := nsq.NewConfig()
	p, _ := nsq.NewProducer("192.168.1.125:4150", config)

	body := make([]byte, 512)
	startCh := make(chan struct{})
	var wg sync.WaitGroup
	parallel := runtime.GOMAXPROCS(0)
	b.Log(parallel)
	b.N = 100000
	doneCh := make(chan *nsq.ProducerTransaction, parallel)
	wg.Add(b.N)

	for j := 0; j < parallel; j++ {
		go func() {
			<-startCh
			for i := 0; i < b.N/parallel; i++ {
				p.PublishAsync("test", body, doneCh)
			}
		}()

		go func() {
			for _ = range doneCh {
				wg.Done()
			}
		}()
	}

	b.StartTimer()
	close(startCh)
	wg.Wait()
}

nsqd config:

nsqd --lookupd-tcp-address=0.0.0.0:4160 --snappy=true --max-msg-size=67108864 --max-msg-timeout=1h --mem-queue-size=100000 

This request does two things, it centralizes logging to the log.go to remove some code duplication with respect to logging. It also make the log an interface so that people can easily drop their own logging structures in.

The structure simply needs to implement the following methods:

    Debugf(format string, v ...interface{})
    Infof(format string, v ...interface{})
    Warnf(format string, v ...interface{})
    Errorf(format string, v ...interface{})

Currently, the ConnectToNSQLookupd method on Consumer fails to return any error when invoked with an invalid nsqlookupd address. Instead the error is logged (and swallowed) in the queryLookupd func, right after the call to apiRequestNegotiateV1, since queryLookupd doesn't return anything. This leaves a consumer with no reliable way to tell that the lookupd address provided was invalid.

To repro:

func foo() {
    q, _ := nsq.NewConsumer("the-topic", "the-channel", nsq.NewConfig())
    handler := &SomeHandler{}
    q.AddHandler(handler)
    err := q.ConnectToNSQLookupd("1.2.3.4:9999")
    // err is nil!
}

We're calling Stop on a NSQ producer, and we're getting the following output after a number of MultiPublishes. Is this io.EOF error expected? We're using the tag 1.0.6.

INFO[0011] INF    1 stopping
INFO[0011] ERR    1 (127.0.0.1:4150) IO error - EOF
INFO[0011] INF    1 (127.0.0.1:4150) beginning close
INFO[0011] INF    1 (127.0.0.1:4150) readLoop exiting

Hey there, I'm using this code to make a producer that sends messages to NSQ. I'm using PublishAsync(), with a "responder" that basically does nothing:

var respChan chan *nsq.ProducerTransaction

func responder(respChan chan *nsq.ProducerTransaction) {
    for _ = range respChan {
        //Get rid of the messages or something... should probably respond, but not sure
        //how to do that yet...
    }
}

I'm sending as many messages as possible, calling publishAsync() in a loop reading from memory. My code maxes out the CPU, and manages about 14,000 messages per second (the messages are quite small, a msgpack'd struct containing a < 16 char string).

In doing a code profile, my code apparently spends 50.5% of its time in "ExternalCode" (no idea what that is).

I've attached my code profile:

Any idea what is going on here?

I got this on Ubuntu 12.04, go 1.2

panic: non-positive interval for NewTicker

goroutine 141 [running]:
runtime.panic(0x815520, 0xc21010d230)
    /usr/local/go/src/pkg/runtime/panic.c:266 +0xb6
time.NewTicker(0x0, 0x3fd8a050fdd8d7f5)
    /usr/local/go/src/pkg/time/tick.go:22 +0x9f
github.com/bitly/go-nsq.(*Consumer).lookupdLoop(0xc21005a8c0)
    /tmp/godep/rev/2b/9a8b5edc036235580256e9a5e5f118a0025151/src/github.com/bitly/go-nsq/consumer.go:228 +0x6e
created by github.com/bitly/go-nsq.(*Consumer).ConnectToNSQLookupd
    /tmp/godep/rev/2b/9a8b5edc036235580256e9a5e5f118a0025151/src/github.com/bitly/go-nsq/consumer.go:216 +0x282

This is https://github.com/bitly/go-nsq/commit/2b9a8b5edc036235580256e9a5e5f118a0025151

I'm connecting to 8 nsqlookupd hosts (which might increase the chance of this happening)

I'm not setting config.lookupdPollInterval in my code.

Right now, in the event that nsqd addresses change, the old connections stay around in the consumer's connections map. This simply cleans up old connections from that map.

the nsq service is deployed on the public network. (106.75.49.135:4161, 106.75.49.135:4150),Can send and receive messages normally, but there will be 'error connecting to nsqd - dial tcp 127.0.0.1:4150: connect: connection refused? '

• docker-compose :

  nsqlookupd:
    image: nsqio/nsq
    command: /nsqlookupd
    ports:
      - "4160:4160"
      - "4161:4161"
  nsqd:
    image: nsqio/nsq
    command:  /nsqd --broadcast-address=nsqd --lookupd-tcp-address=nsqlookupd:4160
    depends_on:
      - nsqlookupd
    ports:
      - "4151:4151"
      - "4150:4150"

• Error message:

2/10/12 11:21:15 INF    1 [Study/Chegg] (nsqd:4150) connecting to nsqd
2022/10/12 11:21:15 ERR    1 [Study/Chegg] (nsqd:4150) error connecting to nsqd - dial tcp 127.0.0.1:4150: connect: connection refused
{"Address":["106.75.49.141:4150"],"level":"info","msg":"Receiving NSQ messages from addresses","time":"2022-10-12T11:21:15+08:00"}

send message:

sub message:

I have a system where I'm getting log messages from nsqd such as:

INFO: PROTOCOL(V2): [127.0.0.1:53376] exiting ioloop ERROR: client(127.0.0.1:53376) - failed to read command - read tcp 1270.0.1:4150->127.0.0.1:53376: i/o timeout INFO: PROTOCOL(v) [127.0.0.1:53376] exiting messagePump

From looking at the /stats endpoint I can see after this message that the consumer is no longer connected (two producers are still connected fine). I can see that there were 2 messages in timeout which I presume is related to the disconnection of the consumer.

However, there is no indication that the consumer knows that it has lost its connection so there's no attempt to reconnect (or even just notify me that it's stopped). The StopChan isn't being called and the Stats() end point on the consumer is still giving a connection count of 1.

Is this the correct behaviour? Have I missed something around how to detect this i/o timeout situation? Unfortunately I can't share the source code, but happy to try things and report back.

Thanks

Andre

Been playing around with NSQ a lot lately and I keep hitting walls when trying to write test suites for assembling various network topologies. Most of the issues seems to stem from NSQD not handling properly consumer disconnects (I'm using go-nsq). I don't even know where to describe the strange things:

• When stopping a consumer, sometimes the CLS message is sent to NSQD, sometimes it is not.
• Even if the CLS does get to NSQD, sometimes it seems to not respond with CLOSE_WAIT, rather nukes the stream.
• The logs are full of error messages on both consumer and broker side during shutdowns that one side or another tries to read/write but the stream is already dead (no graceful disconnect).
• Disconnecting the last consumer doesn't seem to decrement the client count of a topic/channel.
• Disconnecting a consumer doesn't seem to abort/reschedule the in-flight messages for that consumer.

Seems to me that the entire shutdown pathway is very very wrong, just that various timeouts hack around the root cause. E.g. the client heartbeats (or lack thereof after a disconnect) is the one that will trigger the cleanup of leftover client counts; the in-flight timeout is the one that reschedules messages nor processed by a disconnected client.

I'm unsure if I'm doing something weird here, but it seems that NSQ is very very prone to weird behavior when I have very short lived connections.

Is it possible to use singleton NSQ Producer? I'm running an API with NSQ Producer and it seems like i have to open and close connection everytime i hit endpoint.