Feature Request: Edge-Side Buffering for StreamPipes Adapters

[amanchaudhary-95]

Summary

Implement an edge-side buffering mechanism in Apache StreamPipes so that data collected from field devices (e.g., PLCs) is not lost when network connectivity to the remote StreamPipes backend is interrupted.

Problem Statement

In many industrial and IoT scenarios, StreamPipes is deployed at the edge, collecting data from PLCs via adapters and forwarding it to a remote central StreamPipes instance.
However, if the communication between edge → remote breaks (network outage, WAN failure, server maintenance, etc.), all in-flight data is lost because the edge installation currently does not buffer data locally.

This creates reliability issues in environments where continuous, lossless telemetry is critical.

Requested Feature

• Add a local buffer/queue mechanism at the edge StreamPipes instance that:
• Stores incoming events from adapters (e.g., PLC adapter) on local disk or durable queue.
• Automatically retries sending buffered data when the remote system becomes available.
• Flushes the buffer in order (FIFO) to maintain event consistency.

Allows configurable policies:

• max buffer size
• retention period
• retry/backoff strategy
• fallback to “drop oldest” or “stop processing” modes

Why This Is Needed

• Industrial connectivity is not always reliable.
• A temporary WAN outage should not cause data loss.
• Many IoT/edge platforms support offline buffering (e.g., MQTT brokers, Kafka edge nodes, Telegraf, EMQX).
• Adding this would significantly improve StreamPipes’ reliability in edge → cloud deployments.

[ottlukas]

A good opportunity to leverage Apache TSfile https://tsfile.apache.org/ AS local storage. IMHO. https://iotdb.apache.org/

[tenthe]

Thanks for bringing up this idea — I think adding an edge-side buffering mechanism is a great idea, but we should also consider the implications it might have on downstream functionalities

A few initial thoughts:

1. Where should the buffering take place?
I can think of two different levels:

• Client/Broker level:
  A local MQTT/Kafka client could persist messages and automatically resend them once the connection to the remote system is restored.

• StreamPipes adapter level:
  Implementing buffering directly inside the StreamPipes adapter independent of the used protocol

Personally, I think the second approach is better, as we are more flexible when we integrate the functionality ourselves.
Open question: What happens in the current implementation if the connection is interrupted? Does the behavior differ from the protocol used?

2. Handling out-of-order events in pipelines and functions
Once events are buffered and later resent, pipelines and functions may receive data out of chronological order.
Currently, it is generally assumed that events arrive time-synchronized or at least in order. For many functionalities this may not be an issue, but we should identify which ones could be affected.

So we should determine:

• which StreamPipes functionalities are sensitive to out-of-order events,
• how much time/data we want to buffer,
• and whether we need serialization of events or if keeping them in memory is sufficient.

What do you think?
Are there any other implications we should consider?
And do you have suggestions on where a good starting point for the implementation would be?

[amanchaudhary-95]

@tenthe Let me explain my current setup.

I'm running two independent StreamPipes instances on two separate edge systems.
Both StreamPipes instances connect to the same PLC, and together they act as an HA / failover pair.

At the edge, each StreamPipes instance sends data to Kafka, which I use as an edge buffer. Kafka persists messages but cannot automatically forward data to the remote system, so I'm using Apache Flink on the remote side to pull messages from both Kafka topics and write them into my remote time-series database (IoTDB).
At the edge, I am also de-duplicating the messages before they are finally pushed to IoTDB.
Below are screenshots from one of the edge StreamPipes systems:

At the Edge

• StreamPipes adapter reading from PLC

• Pipeline element sending the events into Kafka

However, I believe your suggestion:

StreamPipes adapter level:
Implement buffering directly inside the StreamPipes adapter independent of the protocol used.

…would indeed be a much better long-term solution.

Answer to your first question:

What happens in the current implementation if the connection is interrupted? Does the behavior differ based on the protocol?

Right now, the adapter does not store data locally. If the connection breaks, the data is lost permanently.
So yes, this behavior is independent of the protocol — the adapter has no built-in persistence.

Answer to your second question:

Handling out-of-order events in pipelines and functions.

Normally the sensor data arrives in order, but this is still a real concern. In my current setup, Kafka guarantees message order per partition, so offsets ensure that events are processed correctly and consistently.
If we implement buffering at the StreamPipes adapter level, we can simply add a timestamp tag (if the PLC does not already provide one). This timestamp can then be used to handle out-of-order events during replay or buffering.

I'll submit a separate idea to implement streampipes in cluster so that 2 or more streampipes can act in tandem and can act as HA/Failover.