error while running on cluster

gistfile1.txt

Caused by: java.lang.Exception: Could not complete snapshot 1 for operator Map -> Sink: flink.processor (1/1).
	at org.apache.flink.streaming.api.operators.AbstractStreamOperator.snapshotState(AbstractStreamOperator.java:407)
	at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.checkpointStreamOperator(StreamTask.java:1163)
	at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOperation.executeCheckpointing(StreamTask.java:1095)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState(StreamTask.java:654)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoint(StreamTask.java:590)
	at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpointOnBarrier(StreamTask.java:543)
	... 8 more
Caused by: com.esotericsoftware.kryo.KryoException: java.util.ConcurrentModificationException
Serialization trace:
classes (sun.misc.Launcher$AppClassLoader)
classloader (java.security.ProtectionDomain)
cachedPDs (javax.security.auth.SubjectDomainCombiner)
combiner (java.security.AccessControlContext)
acc (org.apache.flink.runtime.execution.librarycache.FlinkUserCodeClassLoader)
classLoader (org.apache.hadoop.conf.Configuration)
conf (org.apache.flink.runtime.fs.hdfs.HadoopFileSystem)
unsafeFileSystem (org.apache.flink.core.fs.SafetyNetWrapperFileSystem)
fs (com.org.flink.sink.CustomSink)
$outer (com.org.flink.sink.CustomSink$Bucket)
	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:82)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
	at com.esotericsoftware.kryo.Kryo.writeClassAndObject(Kryo.java:599)
	at com.esotericsoftware.kryo.serializers.MapSerializer.write(MapSerializer.java:88)
	at com.esotericsoftware.kryo.serializers.MapSerializer.write(MapSerializer.java:21)
	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
	at com.esotericsoftware.kryo.Kryo.writeObjectOrNull(Kryo.java:577)
	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:68)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)
	at com.esotericsoftware.kryo.Kryo.writeObject(Kryo.java:523)
	at com.esotericsoftware.kryo.serializers.ObjectField.write(ObjectField.java:61)
	at com.esotericsoftware.kryo.serializers.FieldSerializer.write(FieldSerializer.java:495)