Query sometimes aborts processing the first event.

Question

This query sometimes aborts processing the first event:

//--------------------------------------------------------------------

var windowMinutes = 3;
var hoppingSeconds = 5;
var closeEvents = from message in inputStream
                        where message.Machine == "MYPC"
                        && message.Name == "CPU"
                        select message;
var result = from w in closeEvents.HoppingWindow(
  TimeSpan.FromMinutes(windowMinutes), TimeSpan.FromSeconds(hoppingSeconds),                                                                                    
  HoppingWindowOutputPolicy.ClipToWindowEnd)
 
  select new { Value = w.Avg(e => e.Value) };

//----------------------------------------------------------------------------------------------------

QueryTemplate (ExpressionTree):

<QueryTemplate Name="cep:/Server/Application/EventProcessingRuntimeService/QueryTemplate/974f51ac-7f62-4863-a8ab-8990bb77ef8eQuery" xmlns="http://schemas.microsoft.com/ComplexEventProcessing/2010/01/Metadata">
  <Import Name="inputStream" Type="cep:/Server/Application/EventProcessingRuntimeService/EventType/Btm.Services.EventProcessing.Types.PerformanceCounterEvent%2C%20Btm.Services.EventProcessingService%2C%20Version%3D1.0.0.0%2C%20Culture%3Dneutral%2C%20PublicKeyToken%3Dnull">
    <OutputStream Name="Import.3.0" />
  </Import>
  <Export Name="OutputAdapter">
    <InputStream Name="Aggregate.1.2" />
  </Export>
  <Aggregate Name="Aggregate.1.2">
    <InputStream Name="Select.2.1" />
    <OutputStream Name="Aggregate.1.2" />
    <HoppingWindow>
      <WindowDefinition>
        <Size>PT3M</Size>
        <HopSize>PT5S</HopSize>
        <Alignment>0001-01-01T00:00:00Z</Alignment>
      </WindowDefinition>
      <InputPolicy>
        <Clip Left="true" Right="true" />
      </InputPolicy>
      <OutputPolicy>
        <Clip Type="WindowEnd" />
      </OutputPolicy>
    </HoppingWindow>
    <Avg OutputField="Value">
      <InputField Name="Value" StreamName="Select.2.1" />
    </Avg>
  </Aggregate>
  <Select Name="Select.2.1">
    <InputStream Name="Import.3.0" />
    <OutputStream Name="Select.2.1" />
    <FilterExpression>
      <And>
        <Equal>
          <InputField Name="Machine" StreamName="Import.3.0" />
          <Constant Type="System.String" Nullable="true" Value="GENNE-NB2" />
        </Equal>
        <Equal>
          <InputField Name="Name" StreamName="Import.3.0" />
          <Constant Type="System.String" Nullable="true" Value="CPU" />
        </Equal>
      </And>
    </FilterExpression>
  </Select>
</QueryTemplate>

Error Message within the event flow debugger:

Microsoft.ComplexEventProcessing.Engine.OperatorExecutionException: An exception happened when operator 'Aggregate.1.2' was processing event, check inner exception for more details. ---> System.ArgumentOutOfRangeException: Der addierte oder subtrahierte Wert ergibt eine nicht darstellbare DateTime.
Parametername: value
   bei System.DateTime.AddTicks(Int64 value)
   bei System.DateTime.Add(TimeSpan value)
   bei Microsoft.ComplexEventProcessing.Engine.DateTimeExtensions.AddNoOverflow(DateTime dateTime, TimeSpan timeSpan)
   bei Microsoft.ComplexEventProcessing.Engine.DateTimeExtensions.SubtractThrowIfOverflow(DateTime dateTime, TimeSpan timeSpan)
   bei Microsoft.ComplexEventProcessing.Engine.SynopsisManager.SingleHoppingWindowManager.PreviousWindowVs(DateTime timestamp)
   bei Microsoft.ComplexEventProcessing.Engine.SynopsisManager.SingleHoppingWindowManager.IsPhantomCti(DateTime ctiTimestamp)
   bei Microsoft.ComplexEventProcessing.Engine.SynopsisManager.SingleHoppingWindowManager.OverrideCtiPushBackward(DateTime ctiTimestamp)
   bei Microsoft.ComplexEventProcessing.Engine.ExecutionOperatorWindowBasedOrderPreservingMinus.OnProcessEvent(EventReference& eventReference, Int64 stimulusTicks)
   bei Microsoft.ComplexEventProcessing.Engine.ExecutionUserDefinedModule.OnProcessEvent(EventReference& eventReference, Int64 stimulusTicks)
   bei Microsoft.ComplexEventProcessing.Engine.ExecutionOperatorStateful.DoProcessEvent(EventReference& eventReference, Int64 stimulusTicks)
   bei Microsoft.ComplexEventProcessing.Engine.QueryExecutionOperator.ProcessEvent(Int32 streamNo, EventReference& eventReference, Int64 stimulusTicks, Int64 enqueueSequenceNumber)
   --- Ende der internen Ausnahmestapelüberwachung ---
   bei Microsoft.ComplexEventProcessing.Diagnostics.Exceptions.Throw(Exception exception)
   bei Microsoft.ComplexEventProcessing.Engine.QueryExecutionOperator.ProcessEvent(Int32 streamNo, EventReference& eventReference, Int64 stimulusTicks, Int64 enqueueSequenceNumber)
   bei Microsoft.ComplexEventProcessing.Engine.ExecutionOperatorStateful.ProcessEvent(Int32 streamNo, EventReference& eventReference, Int64 stimulusTicks, Int64 enqueueSequenceNumber)
   bei Microsoft.ComplexEventProcessing.Engine.QuerySegmentInputStrategy.DispatchEvents(SchedulingPolicy policy)
   bei Microsoft.ComplexEventProcessing.Engine.SchedulingPolicy.DispatchEvents()
   bei Microsoft.ComplexEventProcessing.Engine.DataflowTask.OnRun()
   bei Microsoft.ComplexEventProcessing.StreamOS.Task.Run()
   bei Microsoft.ComplexEventProcessing.StreamOS.Scheduler.Main()

Please help to find the reason for that.

Answer 1

Does your input adapter factory implement IDeclareAdvanceTimeProperties and specify that the advance time settings are by item count rather than duration? The DataGenerator sample in the StreamInsight samples does.

If so, this is a known bug in 1.2 (and it's on my list to blog about) when you use a hopping or tumbling window and use IDeclareAdvanceTimeProperties specifying the CTIs by event count. To get around this, you can do 1 of 2 things:

• Add an EnqueueCTI at the beginning of your input adapter before any events are enqueued. This can be any date ... something like January 1, 1900 ... before any possible event that you are actually enqueuing ... would work.
• Change the query so to something like this:

var result = from w in closeEvents
    .AlterEventDuration(e => TimeSpan.FromSeconds(0)) //Gets around the bug 
    .HoppingWindow(
        TimeSpan.FromMinutes(windowMinutes),
        TimeSpan.FromSeconds(hoppingSeconds),
        HoppingWindowOutputPolicy.ClipToWindowEnd);

Also, if you stop and then restart the aborted query, it'll work just fine. This only happens on the initial window.

---

DevBiker (aka J Sawyer)
My Blog
My Bike - Concours 14

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 2

Thank You very much, this has saved me a lot of time!

Yes we use the IDeclareAdvanceTimeProperties interface in order to make the incomming events processed directly.

Answer 3

Make sure that you understand exactly how IDeclareAdvanceTimeProperties works. You can see this blog post for some details that show, using LinqPad, how they work.

Typically, I find it better, whenever possible, to enqueue the CTIs directly from the input adapter. This gives you the most control over how they work. However, some adapters (such as the PI input adapter provided by OSISoft) leave you with no option but to declare the advance time properties either in the query or in a wrapper around the adapter factory.

---

DevBiker (aka J Sawyer)
My Blog
My Bike - Concours 14

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.