[Bug]: Consumer failed after 48 hours

[ajhunyady]

I received the following connection error:

received: 607
received: 607
Traceback (most recent call last):
  File "slack-message.py", line 35, in <module>
    main()
  File "slack-message.py", line 32, in main
    consume()
  File "slack-message.py", line 20, in consume
    for i in consumer.stream(Offset.from_end(50)):
  File "/home/ubuntu/.local/lib/python3.8/site-packages/fluvio/__init__.py", line 109, in __next__
    return Record(self._inner.next())
_fluvio_python.Error: Fluvio socket error
Caused by:
Socket closed

The script reads from fluvio and sends message to slack:

import requests
from fluvio import (Fluvio, Offset)

def send_slack_message(message):
    payload = '{"text": "%s"}' % message
    response = requests.post("https://hooks.slack.com/services/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",
            data=payload)

    print(response.text)

def consume():
    value = "0"

    fluvio = Fluvio.connect()
    print("connected to fluvio")

    consumer = fluvio.partition_consumer("fluvio-stars", 0)
    print("retrieved consumer")

    for i in consumer.stream(Offset.from_end(1)):
        received = i.value_string()
        print("received: %s" % received)

        if value == "":
            value = received
        elif value != received:
            value = received
            send_slack_message(value)


def main():
    consume()

if __name__ == "__main__":
    main()

[tjtelan]

The theory is this issue is a side-effect of #1519. The assumption is that both the producer and consumer were running on the same host. And as the producer used up all of the TCP resources, eventually we would see failure in the consumer.

I was able to set up and run some lightly modified example code, which ran concurrently with code from #1519. Both have been running for more than 48 hours

#!/usr/bin/env python3

import requests
import time
from fluvio import (Fluvio, Offset)

def send_slack_message(message):
    payload = '{"text": "%s"}' % message
    #endpoint = "https://hooks.slack.com/services/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"
    endpoint = "https://ptsv2.com/t/op9bn-1630810794/post"
    response = requests.post(endpoint,
            data=payload)

    print(response.text)

def consume():
    value = "0"

    fluvio = Fluvio.connect()
    print("connected to fluvio")

    consumer = fluvio.partition_consumer("fluvio-stars", 0)
    print("retrieved consumer")

    for i in consumer.stream(Offset.from_end(1)):
        received = i.value_string()
        print("received: %s" % received)

        if value == "":
            value = received
        elif value != received:
            value = received
            send_slack_message(value)

        time.sleep(1)


def main():
    consume()

if __name__ == "__main__":
    main()

[tjtelan]

Closing this as a duplicate. The root cause of this issue is tracked by #770 and is a high priority. The client should try to re-connect when the socket closes (like for network-stability reasons)

Let's create a new issue if we see this behavior after #770 is fixed.