search

hortonworks / streamline

StreamLine - Streaming Analytics
Apache License 2.0
164 stars 96 forks source link

Join processor with processing time in test mode doesn't work #1065

Closed HeartSaVioR closed 6 years ago

HeartSaVioR commented 6 years ago

Error message is below:

Exception in thread "main" java.lang.reflect.InvocationTargetException
    at sun.reflect.NativeConstructorAccessorImpl.newInstance0(Native Method)
    at sun.reflect.NativeConstructorAccessorImpl.newInstance(NativeConstructorAccessorImpl.java:62)
    at sun.reflect.DelegatingConstructorAccessorImpl.newInstance(DelegatingConstructorAccessorImpl.java:45)
    at java.lang.reflect.Constructor.newInstance(Constructor.java:422)
    at org.apache.storm.flux.FluxBuilder.buildObject(FluxBuilder.java:310)
    at org.apache.storm.flux.FluxBuilder.buildComponents(FluxBuilder.java:358)
    at org.apache.storm.flux.FluxBuilder.buildTopology(FluxBuilder.java:74)
    at org.apache.storm.flux.Flux.runCli(Flux.java:153)
    at org.apache.storm.flux.Flux.main(Flux.java:98)
Caused by: com.fasterxml.jackson.streamline.databind.JsonMappingException: null tsFields
 at [Source: {"windowLength":{"class":".Window$Duration","durationMs":1000},"slidingInterval":{"class":".Window$Duration","durationMs":1000},"tsFields":null,"lagMs":0}; line: 1, column: 140] (through reference chain: com.hortonworks.streamline.streams.layout.component.rule.expression.Window["tsFields"])
    at com.fasterxml.jackson.streamline.databind.JsonMappingException.from(JsonMappingException.java:223)
    at com.fasterxml.jackson.streamline.databind.deser.SettableBeanProperty._throwAsIOE(SettableBeanProperty.java:537)
    at com.fasterxml.jackson.streamline.databind.deser.SettableBeanProperty._throwAsIOE(SettableBeanProperty.java:518)
    at com.fasterxml.jackson.streamline.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:99)
    at com.fasterxml.jackson.streamline.databind.deser.BeanDeserializer.vanillaDeserialize(BeanDeserializer.java:260)
    at com.fasterxml.jackson.streamline.databind.deser.BeanDeserializer.deserialize(BeanDeserializer.java:125)
    at com.fasterxml.jackson.streamline.databind.ObjectMapper._readMapAndClose(ObjectMapper.java:3789)
    at com.fasterxml.jackson.streamline.databind.ObjectMapper.readValue(ObjectMapper.java:2779)
    at com.hortonworks.streamline.streams.layout.component.rule.expression.Window.<init>(Window.java:126)
    ... 9 more
Caused by: java.lang.NullPointerException: null tsFields
    at java.util.Objects.requireNonNull(Objects.java:228)
    at com.hortonworks.streamline.streams.layout.component.rule.expression.Window.setTsFields(Window.java:168)
    at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method)
    at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:497)
    at com.fasterxml.jackson.streamline.databind.deser.impl.MethodProperty.deserializeAndSet(MethodProperty.java:97)
    ... 14 more

(Strange behavior observed is that stack trace information was not provided unless I changed the log level to DEBUG.)

Relevant component definition in yaml file is below:

- id: ruleBolt4cc2a3af-f012-4e35-9ba1-017d5196a3a1
  className: com.hortonworks.streamline.streams.runtime.storm.bolt.rules.RulesBolt
  constructorArgs:
  - '{"id":"4","topologyComponentBundleId":"4","name":"RULE","config":{"properties":{"rules":[3]}},"transformationClass":"com.hortonworks.streamline.streams.layout.storm.RuleBoltFluxComponent","outputStreams":[{"id":"rule_transform_stream_3","schema":{"fields":[{"name":"driverId","type":"INTEGER","optional":false},{"name":"truckId","type":"INTEGER","optional":false},{"name":"eventTime","type":"STRING","optional":false}]}},{"id":"rule_notifier_stream_3","schema":{"fields":[{"name":"driverId","type":"INTEGER","optional":false},{"name":"truckId","type":"INTEGER","optional":false},{"name":"eventTime","type":"STRING","optional":false}]}}],"rules":[{"id":3,"name":"r1","description":"sds","ruleProcessorName":null,"streams":["join_processor_stream_11"],"projection":null,"condition":{"expression":{"class":"com.hortonworks.streamline.streams.layout.component.rule.expression.BinaryExpression","operator":"GREATER_THAN","first":{"class":"com.hortonworks.streamline.streams.layout.component.rule.expression.FieldExpression","value":{"name":"truckId","type":"INTEGER","optional":false}},"second":{"class":"com.hortonworks.streamline.streams.layout.component.rule.expression.Literal","value":"15"}}},"groupBy":null,"having":null,"window":null,"actions":[{"__type":"com.hortonworks.streamline.streams.layout.component.rule.action.TransformAction","name":"transformAction","outputStreams":["rule_transform_stream_3"],"transforms":[]}],"referredUdfs":[]}],"processAll":true,"extraResources":[],"extraJars":[]}'
  - SQL

This error is caused by #1029. There's guard against null so that it prevents setting null to the list, but the value for tsFields is null anyway if join processor uses processing time, and below validation throws exception while deserializing Window json. (I haven't check it with event time though for now.)

https://github.com/hortonworks/streamline/blob/d95448e1a6ecf72054dd12e9257cfbb1fc3da9b0/streams/layout/src/main/java/com/hortonworks/streamline/streams/layout/component/rule/expression/Window.java#L167-L170

HeartSaVioR commented 6 years ago

Provided fix in #1066 and also provided alternative way to fix in that PR.