Package org.nuxeo.lib.stream.computation.log

Class LogStreamProcessor

java.lang.Object
org.nuxeo.lib.stream.computation.log.LogStreamProcessor
All Implemented Interfaces:
StreamProcessor
public class LogStreamProcessor extends Object implements StreamProcessor
Since:
9.3

  • Field Details

    • manager

      protected final LogManager manager

    • topology

      protected Topology topology

    • settings

      protected Settings settings

    • pools

      protected List<ComputationPool> pools

    • streamManager

      protected LogStreamManager streamManager

    • needRegister

      protected final boolean needRegister

    • OBJECT_MAPPER

      protected static final com.fasterxml.jackson.databind.ObjectMapper OBJECT_MAPPER

  • Constructor Details

  • Method Details

    • init

      public StreamProcessor init(Topology topology, Settings settings)
      Description copied from interface: StreamProcessor
      Initialize streams, but don't run the computations
      Specified by:
      init in interface StreamProcessor

    • start

      public void start()
      Description copied from interface: StreamProcessor
      Run the initialized computations.
      Specified by:
      start in interface StreamProcessor

    • waitForAssignments

      public boolean waitForAssignments(Duration timeout) throws InterruptedException
      Description copied from interface: StreamProcessor
      Wait for the computations to have assigned partitions ready to process records. The processor must be started. This is useful for writing unit test.

      Returns true if all computations have assigned partitions during the timeout delay.

      Specified by:
      waitForAssignments in interface StreamProcessor
      Throws:
      InterruptedException

    • isTerminated

      public boolean isTerminated()
      Description copied from interface: StreamProcessor
      True if there is no active processing threads.
      Specified by:
      isTerminated in interface StreamProcessor

    • toJson

      public String toJson(Map<String,String> meta)
      Description copied from interface: StreamProcessor
      Returns a JSON representation of the processor, this includes the list of streams and computations with their settings along with the topology.
      Specified by:
      toJson in interface StreamProcessor

    • stopComputation

      public boolean stopComputation(Name computation)
      Description copied from interface: StreamProcessor
      Stop a computation thread pool.
      Specified by:
      stopComputation in interface StreamProcessor
      Returns:
      true if computation thread pool was found and stopped.

    • startComputation

      public boolean startComputation(Name computation)
      Description copied from interface: StreamProcessor
      Start a computation thread pool that has been stopped using StreamProcessor.stopComputation(Name)
      Specified by:
      startComputation in interface StreamProcessor
      Returns:
      true if computation thread pool has been started.

    • getEdgeName

      protected String getEdgeName(Topology.Vertex edge)

    • stop

      public boolean stop(Duration timeout)
      Description copied from interface: StreamProcessor
      Try to stop computations gracefully after processing a record or a timer within the timeout duration. If this can not be done within the timeout, shutdown and returns false.
      Specified by:
      stop in interface StreamProcessor

    • drainAndStop

      public boolean drainAndStop(Duration timeout)
      Description copied from interface: StreamProcessor
      Stop computations when input streams are empty. The timeout is applied for each computation, the total duration can be up to nb computations * timeout

      Returns true if computations are stopped during the timeout delay.

      Specified by:
      drainAndStop in interface StreamProcessor

    • shutdown

      public void shutdown()
      Description copied from interface: StreamProcessor
      Shutdown immediately.
      Specified by:
      shutdown in interface StreamProcessor

    • getLowWatermark

      public long getLowWatermark()
      Description copied from interface: StreamProcessor
      Returns the low watermark for all the computations of the topology. Any message with an offset below the low watermark has been processed. The returned watermark is local to this processing node.
      Specified by:
      getLowWatermark in interface StreamProcessor

    • getLatency

      public Latency getLatency(String computationName)
      Description copied from interface: StreamProcessor
      Returns the latency for a computation. This works also for distributed computations.
      Specified by:
      getLatency in interface StreamProcessor

    • getLowWatermark

      public long getLowWatermark(String computationName)
      Description copied from interface: StreamProcessor
      Returns the low watermark for the computation. Any message with an offset below the low watermark has been processed by this computation and its ancestors. The returned watermark is local to this processing node, if the computation is distributed the global low watermark is the minimum of all nodes low watermark.
      Specified by:
      getLowWatermark in interface StreamProcessor

    • isDone

      public boolean isDone(long timestamp)
      Description copied from interface: StreamProcessor
      Returns true if all messages with a lower timestamp has been processed by the topology.
      Specified by:
      isDone in interface StreamProcessor

    • initPools

      protected List<ComputationPool> initPools()

    • getCodecForStreams

      protected Codec<Record> getCodecForStreams(String name, Set<String> streams)

    • getDefaultAssignments

      protected List<List<LogPartition>> getDefaultAssignments(ComputationMetadataMapping meta)