diff --git a/plugins/inputs/mqtt_consumer/mqtt_consumer.go b/plugins/inputs/mqtt_consumer/mqtt_consumer.go
index 5853ad93..85b7074a 100644
--- a/plugins/inputs/mqtt_consumer/mqtt_consumer.go
+++ b/plugins/inputs/mqtt_consumer/mqtt_consumer.go
@@ -150,27 +150,25 @@ func (m *MQTTConsumer) connect() error {
                return err
        }

-       go m.receiver()
-
-       return nil
-}
-
-func (m *MQTTConsumer) onConnect(c mqtt.Client) {
        log.Printf("I! MQTT Client Connected")
-       if !m.PersistentSession || !m.connected {
+       m.connected = true
+
+       if !m.PersistentSession {
                topics := make(map[string]byte)
                for _, topic := range m.Topics {
                        topics[topic] = byte(m.QoS)
                }
-               subscribeToken := c.SubscribeMultiple(topics, m.recvMessage)
+               subscribeToken := m.client.SubscribeMultiple(topics, m.recvMessage)
                subscribeToken.Wait()
                if subscribeToken.Error() != nil {
                        m.acc.AddError(fmt.Errorf("E! MQTT Subscribe Error\ntopics: %s\nerror: %s",
                                strings.Join(m.Topics[:], ","), subscribeToken.Error()))
                }
-               m.connected = true
        }
-       return
+
+       go m.receiver()
+
+       return nil
 }
 
 func (m *MQTTConsumer) onConnectionLost(c mqtt.Client, err error) {
@@ -271,10 +269,9 @@ func (m *MQTTConsumer) createOpts() (*mqtt.ClientOptions, error) {
                
                opts.AddBroker(server)
        }
-       opts.SetAutoReconnect(true)
+       opts.SetAutoReconnect(false)
        opts.SetKeepAlive(time.Second * 60)
        opts.SetCleanSession(!m.PersistentSession)
-       opts.SetOnConnectHandler(m.onConnect)
        opts.SetConnectionLostHandler(m.onConnectionLost)

        return opts, nil