The last one is C or A?
How often is it failing (every nth run?) Is it always failing at the same
execute() call, or at different ones?
Is it always the exact same exception or is it different ones?
Does the error behave differently depending on the input data?
Sorry for asking so many questions, but these errors can have many causes
and just searching the code for potential issues can take a lot of time ;)
On Tue, May 17, 2016 at 12:47 PM, Flavio Pompermaier <pomperma...@okkam.it>
wrote:
> Ah sorry, I forgot to mention that I don't use any custom kryo
> serializers..
>
> On Tue, May 17, 2016 at 12:39 PM, Flavio Pompermaier <pomperma...@okkam.it
> > wrote:
>
>> I got those exceptions running 3 different types of jobs..I could have
>> tracked the job and the error...my bad!
>> However, the most problematic job is the last one, where I run a series
>> of jobs one after the other (calling env.execute() in a for loop)..
>> I you want I can share with you my code (in private for the moment
>> because it's not public yet) or the dashboard screen via skype while the
>> jobs are running..
>>
>>
>> On Tue, May 17, 2016 at 12:31 PM, Robert Metzger <rmetz...@apache.org>
>> wrote:
>>
>>> Hi Flavio,
>>> thank you for providing additional details.
>>> I don't think that missing hashCode / equals() implementations cause
>>> such an error. They can cause wrong sorting or partitioning of the data,
>>> but the serialization should still work properly.
>>> I suspect the issue somewhere in the serialization stack.
>>>
>>> Are you registering any custom kryo serializers?
>>>
>>>
>>>
>>> From your past emails, you saw the following different exceptions:
>>>
>>> A) Caused by: java.io.UTFDataFormatException: malformed input around
>>> byte 42
>>> B) Caused by: java.lang.ClassNotFoundException: java.ttil.HashSet
>>> C) Caused by: java.lang.ArrayIndexOutOfBoundsException: -2
>>>
>>> Were they all caused by the same job, or different ones?
>>>
>>>
>>> On Tue, May 17, 2016 at 12:12 PM, Flavio Pompermaier <
>>> pomperma...@okkam.it> wrote:
>>>
>>>> Hi Robert,
>>>> in this specific case the interested classes are:
>>>>
>>>> - Tuple3<String, String, IndexAttributeToExpand>
>>>> (IndexAttributeToExpand is a POJO extending another class and both of
>>>> them
>>>> doesn't implement equals and hashcode)
>>>> - Tuple3<String,String, TreeNode<String, Map<String, Set<String>>>>
>>>> (TreeNode is a POJO containing other TreeNode and it doesn't
>>>> implement equals and hashcode)
>>>> - Now I've added to both classes hashCode and equals in order to be
>>>> aligned with POJO policies, however the job finished correctly after
>>>> stopping and restarting the cluster...usually when I have strange
>>>> serialization exception I stop and restart the cluster and everything
>>>> works.
>>>>
>>>> The TreeNode class (the recursive one) is actually the following:
>>>>
>>>> import java.io.Serializable;
>>>> import java.util.ArrayList;
>>>> import java.util.HashMap;
>>>> import java.util.Iterator;
>>>> import java.util.LinkedList;
>>>> import java.util.List;
>>>> import java.util.UUID;
>>>>
>>>> import com.fasterxml.jackson.annotation.JsonIgnore;
>>>>
>>>> public class TreeNode<K,V> implements Serializable {
>>>>
>>>> private static final long serialVersionUID = 1L;
>>>> private int level = 0;
>>>>
>>>> private String uuid;
>>>> private K key;
>>>> private V value;
>>>> @JsonIgnore
>>>> private TreeNode<K,V> parent;
>>>> private List<TreeNode<K,V>> children;
>>>> @JsonIgnore
>>>> private HashMap<K, List<TreeNode<K,V>>> lookup;
>>>>
>>>> public TreeNode(K key, V value) {
>>>> this.level = 0;
>>>> this.key = key;
>>>> this.uuid = UUID.randomUUID().toString();
>>>> this.value = value;
>>>> this.children = new LinkedList<TreeNode<K,V>>();
>>>> List<TreeNode<K, V>> thisAsList = new
>>>> ArrayList<TreeNode<K,V>>();
>>>> thisAsList.add(this);
>>>> this.lookup = new HashMap<K, List<TreeNode<K,V>>>();
>>>> this.lookup.put(key, thisAsList);
>>>> }
>>>>
>>>> public TreeNode<K,V> addChild(K key, V value) {
>>>> TreeNode<K,V> childNode = new TreeNode<K,V>(key, value);
>>>> childNode.level = level +1;
>>>> childNode.parent = this;
>>>> childNode.lookup = lookup;
>>>> childNode.uuid = UUID.randomUUID().toString();
>>>> this.children.add(childNode);
>>>> List<TreeNode<K, V>> l = lookup.get(childNode.key);
>>>> if(l==null){
>>>> l = new ArrayList<TreeNode<K,V>>();
>>>> lookup.put(childNode.key, l);
>>>> }
>>>> l.add(childNode);
>>>> return childNode;
>>>> }
>>>>
>>>> public boolean isLeaf() {
>>>> return children.isEmpty() ;
>>>> }
>>>> public int getLevel() {
>>>> return level;
>>>> }
>>>> public TreeNode<K,V> getParent() {
>>>> return parent;
>>>> }
>>>> public V getValue() {
>>>> return value;
>>>> }
>>>> public String getUuid() {
>>>> return uuid;
>>>> }
>>>> public void setUuid(String uuid) {
>>>> this.uuid = uuid;
>>>> }
>>>> public List<TreeNode<K,V>> getChildren() {
>>>> return children;
>>>> }
>>>> public List<TreeNode<K, V>> getNodesByKey(K key) {
>>>> return lookup.get(key);
>>>> }
>>>> public K getKey() {
>>>> return key;
>>>> }
>>>> public List<TreeNode<K,V>> getLeafs() {
>>>> List<TreeNode<K,V>> ret = new ArrayList<TreeNode<K,V>>();
>>>> getLeafs(ret);
>>>> return ret;
>>>> }
>>>> private void getLeafs(List<TreeNode<K, V>> ret) {
>>>> if(children.isEmpty())
>>>> ret.add(this);
>>>> for (TreeNode<K, V> child : children) {
>>>> child.getLeafs(ret);
>>>> }
>>>> }
>>>>
>>>> @Override
>>>> public String toString() {
>>>> return toString(true);
>>>> }
>>>> public String toString(boolean withChildren) {
>>>> if(key==null)
>>>> return super.toString();
>>>> StringBuffer ret = new StringBuffer();
>>>> for (int i = 0; i < level; i++) {
>>>> ret.append(" >");
>>>> }
>>>> ret.append(" " +key.toString());
>>>> if(withChildren){
>>>> for (TreeNode<K, V> child : children) {
>>>> ret.append("\n").append(child.toString());
>>>> }
>>>> }
>>>> return ret.toString();
>>>> }
>>>>
>>>> public void setValue(V value) {
>>>> this.value = value;
>>>> }
>>>>
>>>> public void remove(List<TreeNode<K, V>> nodes) {
>>>> for (TreeNode<K, V> n : nodes) {
>>>> removeChildren(n);
>>>> }
>>>> for (TreeNode<K, V> n : nodes) {
>>>> TreeNode<K, V> parent = n.getParent();
>>>> if(parent==null)
>>>> continue;
>>>> parent.children.remove(n);
>>>> }
>>>> }
>>>>
>>>> private void removeChildren(TreeNode<K, V> node) {
>>>> lookup.remove(node.getUuid());
>>>> if(node.children.isEmpty())
>>>> return;
>>>> Iterator<TreeNode<K, V>> it = node.children.iterator();
>>>> while (it.hasNext()) {
>>>> TreeNode<K, V> child = (TreeNode<K, V>) it.next();
>>>> removeChildren(child);
>>>> it.remove();
>>>> }
>>>> }
>>>>
>>>> public void clear() {
>>>> this.key = null;
>>>> this.value = null;
>>>> this.uuid = null;
>>>> for (TreeNode<K, V> child : children) {
>>>> child.clear();
>>>> }
>>>> this.children.clear();
>>>> this.lookup.clear();
>>>> }
>>>>
>>>> public TreeNode<K, V> getNodeById(K key, String uuid) {
>>>> List<TreeNode<K, V>> nodes = getNodesByKey(key);
>>>> for (TreeNode<K, V> treeNode : nodes) {
>>>> if(uuid.equals(treeNode.getUuid()))
>>>> return treeNode;
>>>> }
>>>> return null;
>>>> }
>>>>
>>>> public HashMap<K, List<TreeNode<K, V>>> getLookup() {
>>>> return lookup;
>>>> }
>>>>
>>>> public void setLookup(HashMap<K, List<TreeNode<K, V>>> lookup) {
>>>> this.lookup = lookup;
>>>> }
>>>>
>>>> public void setLevel(int level) {
>>>> this.level = level;
>>>> }
>>>>
>>>> public void setKey(K key) {
>>>> this.key = key;
>>>> }
>>>>
>>>> public void setParent(TreeNode<K, V> parent) {
>>>> this.parent = parent;
>>>> }
>>>>
>>>> public void setChildren(List<TreeNode<K, V>> children) {
>>>> this.children = children;
>>>> }
>>>> }
>>>>
>>>> On Tue, May 17, 2016 at 12:00 PM, Robert Metzger <rmetz...@apache.org>
>>>> wrote:
>>>>
>>>>> Hi Flavio,
>>>>>
>>>>> which datatype are you using?
>>>>>
>>>>> On Tue, May 17, 2016 at 11:42 AM, Flavio Pompermaier <
>>>>> pomperma...@okkam.it> wrote:
>>>>>
>>>>>> Hi to all,
>>>>>> during these days we've run a lot of Flink jobs and from time to time
>>>>>> (apparently randomly) a different Exception arise during their
>>>>>> executions...
>>>>>> I hope one of them could help in finding the source of the
>>>>>> problem..This time the exception is:
>>>>>>
>>>>>> An error occurred while reading the next record.
>>>>>> at
>>>>>> org.apache.flink.runtime.util.NonReusingKeyGroupedIterator.advanceToNext(NonReusingKeyGroupedIterator.java:148)
>>>>>> at
>>>>>> org.apache.flink.runtime.util.NonReusingKeyGroupedIterator.access$300(NonReusingKeyGroupedIterator.java:32)
>>>>>> at
>>>>>> org.apache.flink.runtime.util.NonReusingKeyGroupedIterator$ValuesIterator.next(NonReusingKeyGroupedIterator.java:192)
>>>>>> at
>>>>>> org.okkam.entitons.mapping.flink.IndexMappingExecutor$TupleToEntitonJsonNode.reduce(IndexMappingExecutor.java:61)
>>>>>> at
>>>>>> org.apache.flink.runtime.operators.GroupReduceDriver.run(GroupReduceDriver.java:125)
>>>>>> at
>>>>>> org.apache.flink.runtime.operators.BatchTask.run(BatchTask.java:480)
>>>>>> at
>>>>>> org.apache.flink.runtime.operators.BatchTask.invoke(BatchTask.java:345)
>>>>>> at org.apache.flink.runtime.taskmanager.Task.run(Task.java:559)
>>>>>> at java.lang.Thread.run(Thread.java:745)
>>>>>> Caused by: java.io.UTFDataFormatException: malformed input around
>>>>>> byte 42
>>>>>> at
>>>>>> org.apache.flink.runtime.memory.AbstractPagedInputView.readUTF(AbstractPagedInputView.java:488)
>>>>>> at
>>>>>> org.apache.flink.api.java.typeutils.runtime.PojoSerializer.deserialize(PojoSerializer.java:403)
>>>>>> at
>>>>>> org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:135)
>>>>>> at
>>>>>> org.apache.flink.api.java.typeutils.runtime.TupleSerializer.deserialize(TupleSerializer.java:30)
>>>>>> at
>>>>>> org.apache.flink.runtime.io.disk.ChannelReaderInputViewIterator.next(ChannelReaderInputViewIterator.java:100)
>>>>>> at
>>>>>> org.apache.flink.runtime.operators.sort.MergeIterator$HeadStream.nextHead(MergeIterator.java:161)
>>>>>> at
>>>>>> org.apache.flink.runtime.operators.sort.MergeIterator.next(MergeIterator.java:113)
>>>>>> at
>>>>>> org.apache.flink.runtime.util.NonReusingKeyGroupedIterator.advanceToNext(NonReusingKeyGroupedIterator.java:130
>>>>>>
>>>>>> Could this error be cause by a missing implementation of hashCode()
>>>>>> and equals()?
>>>>>>
>>>>>> Thanks in advance,
>>>>>> Flavio
>>>>>>
>>>>>
>>>>
>>>
>>
