It happens when it looks to throw an exception and calls shardBuffer.toString. b'coz of the check....
int id = sharedBuffer.entryId; Preconditions.checkState(id != -1, "Could not find id for entry: " + *sharedBuffer*); On Thu, Feb 1, 2018 at 5:09 PM, Vishal Santoshi <vishal.santo...@gmail.com> wrote: > The watermark has not moved for this pattern to succeed ( or other wise ), > the issue though is that it is pretty early in the pipe ( like within a > minute ). I am replaying from a kafka topic but the keyed operator has > emitted no more than 1500 plus elements to SelectCEPOperator ( very visible > on the UI ) so am sure not enough elements have been added to the > SharedBuffer to create memory stress. > > The nature of the input stream is that events are pushed out with a > specific timestamp ( it is a time series and the timestamp if the beginning > of the time slot ) as in one will have a bunch of elements that have a > constant timestamp till the next batch appears. > > A batch though does not have more than the number of keys elements ( 600 > ). > > On Thu, Feb 1, 2018 at 4:53 PM, Vishal Santoshi <vishal.santo...@gmail.com > > wrote: > >> This is a pretty simple pattern, as in I hardly have 1500 elements ( across >> 600 keys at the max ) put in >> >> and though I have a pretty wide range , as in I am looking at a relaxed >> pattern ( like 40 true conditions in 6 hours ), >> >> I get this. I have the EventTime turned on. >> >> >> >> java.lang.OutOfMemoryError: Java heap space >> >> at java.util.Arrays.copyOf(Arrays.java:3332) >> at >> java.lang.AbstractStringBuilder.ensureCapacityInternal(AbstractStringBuilder.java:124) >> at >> java.lang.AbstractStringBuilder.append(AbstractStringBuilder.java:448) >> at java.lang.StringBuilder.append(StringBuilder.java:136) >> at java.lang.StringBuilder.append(StringBuilder.java:131) >> at org.apache.commons.lang3.StringUtils.join(StringUtils.java:4106) >> at org.apache.commons.lang3.StringUtils.join(StringUtils.java:4151) >> at >> org.apache.flink.cep.nfa.SharedBuffer$SharedBufferEntry.toString(SharedBuffer.java:624) >> at java.lang.String.valueOf(String.java:2994) >> at java.lang.StringBuilder.append(StringBuilder.java:131) >> at >> org.apache.flink.cep.nfa.SharedBuffer$SharedBufferSerializer.serialize(SharedBuffer.java:964) >> at >> org.apache.flink.cep.nfa.SharedBuffer$SharedBufferSerializer.serialize(SharedBuffer.java:835) >> at org.apache.flink.cep.nfa.NFA$NFASerializer.serialize(NFA.java:888) >> at org.apache.flink.cep.nfa.NFA$NFASerializer.serialize(NFA.java:820) >> at >> org.apache.flink.contrib.streaming.state.RocksDBValueState.update(RocksDBValueState.java:100) >> >> . >> >> . >> >> . >> >> >> Any one has seen this issue ? >> >> >> >
