[OpenNMS/opennms] 86fefa: Correctness

  Branch: refs/heads/dcy/flow-aggregation
  Home:   https://github.com/OpenNMS/opennms
  Commit: 86fefad2eec1db7bf178952b481299cdcd743b32
      https://github.com/OpenNMS/opennms/commit/86fefad2eec1db7bf178952b481299cdcd743b32
  Author: Dino <[email protected]>
  Date:   2026-07-28 (Tue, 28 Jul 2026)

  Changed paths:
    M features/flows/aggregation/src/main/java/org/opennms/netmgt/flows/aggregation/AbstractAggregatingFlowRepository.java
    M features/flows/aggregation/src/main/java/org/opennms/netmgt/flows/aggregation/AggregatedFlowSink.java
    M features/flows/aggregation/src/main/java/org/opennms/netmgt/flows/aggregation/FixedWindowAggregation.java
    M features/flows/aggregation/src/main/java/org/opennms/netmgt/flows/aggregation/FlowAggregator.java
    M features/flows/aggregation/src/main/java/org/opennms/netmgt/flows/aggregation/FlowInput.java
    M features/flows/aggregation/src/main/java/org/opennms/netmgt/flows/aggregation/FlowProration.java
    M features/flows/aggregation/src/test/java/org/opennms/netmgt/flows/aggregation/FixedWindowAggregationTest.java
    M features/flows/aggregation/src/test/java/org/opennms/netmgt/flows/aggregation/FlowAggregatorTest.java
    M features/flows/aggregation/src/test/java/org/opennms/netmgt/flows/aggregation/ProportionalSumParityTest.java

  Log Message:
  -----------
    Correctness

  - Sink failure / misleading log — flushes now drain under a lock into a batch and hand it to the sink outside the lock via a new deliver() that catches, counts rowsDroppedOnSink, and logs "dropped (not retried)". The AggregatedFlowSink javadoc documents the best-effort contract (a throw drops that batch; the sink owns durability; readers must tolerate a repeated
  (window,key,writer)).
  - Timestamp poisoning — add() rejects a flow (counts flowsDroppedInvalid) when lastSwitched > now + 1h or span > 1 day, before touching the watermark — so one bad record can't strand every later flow, and a near-epoch delta can't spin the window loop.
  - add()/flusher race — a ReentrantReadWriteLock: ingestion takes the read lock, eviction the write lock, so a flow can't land in a window mid-eviction.
  - Re-emission — a lock-guarded emittedFloor (updated in both flush paths) makes any evicted window non-resurrectable, so idle-flush + late arrival no longer double-counts.
  - FlowInput.from — null-checks getBytes() (was an unboxing NPE); UNKNOWN direction now maps to INGRESS + inputSnmp per the OpenNMS convention.

  Hygiene

  - Thread renamed to flow-aggregator; no broken @links / stray backend tokens remain (full sweep clean).
  - AggregatedFlowSink is now real — the ctor and createSink take/return it instead of a raw Consumer.
  - perNodeShift uses Math.floorMod (kills the Integer.MIN_VALUE negative-shift landmine); golden values recomputed and verified.
  - Removed the thinking-out-loud comment, added trailing newlines, and cut the fuzz loops (200k/100k/100k → 20k each; parity test ~2.6s→~1s).
  - Added 7 regression tests covering every correctness fix.



To unsubscribe from these emails, change your notification settings at https://github.com/OpenNMS/opennms/settings/notifications


_______________________________________________
Please read the OpenNMS Mailing List FAQ:
http://www.opennms.org/wiki/index.php?page=MailingListFaq
opennms-cvs mailing list

To *unsubscribe* or change your subscription options, see the bottom of this page:
https://lists.sourceforge.net/lists/listinfo/opennms-cvs