Mailing-List: contact issues-help@flink.apache.org; run by ezmlm
Precedence: bulk
Reply-To: dev@flink.apache.org
Date: Thu, 9 Nov 2017 13:22:00 +0000 (UTC)
From: "ASF GitHub Bot (JIRA)" <jira@apache.org>
To: issues@flink.apache.org
Message-ID: <JIRA.13116624.1510050539000.195539.1510233720142@Atlassian.JIRA>
In-Reply-To: <JIRA.13116624.1510050539000@Atlassian.JIRA>
References: <JIRA.13116624.1510050539000@Atlassian.JIRA> <JIRA.13116624.1510050539218@jira-lw-us.apache.org>
Subject: [jira] [Commented] (FLINK-8005) Snapshotting FlinkKafkaProducer011
 fails due to ClassLoader issues
MIME-Version: 1.0
Content-Type: text/plain; charset=utf-8
Content-Transfer-Encoding: quoted-printable
archived-at: Thu, 09 Nov 2017 13:22:06 -0000


    [ https://issues.apache.org/jira/browse/FLINK-8005?page=3Dcom.atlassian=
.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=3D1624=
5623#comment-16245623 ]=20

ASF GitHub Bot commented on FLINK-8005:
---------------------------------------

Github user aljoscha commented on the issue:

    https://github.com/apache/flink/pull/4980
 =20
    I think waiting on the stop latch might not be enough (in 100 % of case=
s) because the other two calls are also asynchronous.


> Snapshotting FlinkKafkaProducer011 fails due to ClassLoader issues
> ------------------------------------------------------------------
>
>                 Key: FLINK-8005
>                 URL: https://issues.apache.org/jira/browse/FLINK-8005
>             Project: Flink
>          Issue Type: Bug
>          Components: Core, Kafka Connector, State Backends, Checkpointing
>    Affects Versions: 1.4.0
>            Reporter: Gary Yao
>            Assignee: Gary Yao
>            Priority: Blocker
>             Fix For: 1.4.0
>
>
> *Problem Description*
> Classes in the user code jar cannot be loaded by the snapshot thread=E2=
=80=99s context class loader ({{AppClassLoader}}).
> For example, when creating instances of {{KafkaProducer}}, Strings are re=
solved to class objects by Kafka.
> Find below an extract from {{ConfigDef.java}}:
> {code}
> case CLASS:
>     if (value instanceof Class)
>         return value;
>     else if (value instanceof String)
>         return Class.forName(trimmed, true, Utils.getContextOrKafkaClassL=
oader());
>     else
>         throw new ConfigException(name, value, "Expected a Class instance=
 or class name.");
> {code}
> *Exception/Stacktrace*
> {noformat}
> Caused by: java.lang.Exception: Could not complete snapshot 1 for operato=
r Source: Collection Source -> Sink: kafka-sink-1510048188383 (1/1).
> =09at org.apache.flink.streaming.api.operators.AbstractStreamOperator.sna=
pshotState(AbstractStreamOperator.java:379)
> =09at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOp=
eration.checkpointStreamOperator(StreamTask.java:1077)
> =09at org.apache.flink.streaming.runtime.tasks.StreamTask$CheckpointingOp=
eration.executeCheckpointing(StreamTask.java:1026)
> =09at org.apache.flink.streaming.runtime.tasks.StreamTask.checkpointState=
(StreamTask.java:659)
> =09at org.apache.flink.streaming.runtime.tasks.StreamTask.performCheckpoi=
nt(StreamTask.java:595)
> =09at org.apache.flink.streaming.runtime.tasks.StreamTask.triggerCheckpoi=
nt(StreamTask.java:526)
> =09... 7 more
> Caused by: org.apache.kafka.common.config.ConfigException: Invalid value =
org.apache.kafka.common.serialization.ByteArraySerializer for configuration=
 key.serializer: Class org.apache.kafka.common.serialization.ByteArraySeria=
lizer could not be found.
> =09at org.apache.kafka.common.config.ConfigDef.parseType(ConfigDef.java:7=
15)
> =09at org.apache.kafka.common.config.ConfigDef.parseValue(ConfigDef.java:=
460)
> =09at org.apache.kafka.common.config.ConfigDef.parse(ConfigDef.java:453)
> =09at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig=
.java:62)
> =09at org.apache.kafka.common.config.AbstractConfig.<init>(AbstractConfig=
.java:75)
> =09at org.apache.kafka.clients.producer.ProducerConfig.<init>(ProducerCon=
fig.java:360)
> =09at org.apache.kafka.clients.producer.KafkaProducer.<init>(KafkaProduce=
r.java:288)
> =09at org.apache.flink.streaming.connectors.kafka.internal.FlinkKafkaProd=
ucer.<init>(FlinkKafkaProducer.java:114)
> =09at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.i=
nitProducer(FlinkKafkaProducer011.java:913)
> =09at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.i=
nitTransactionalProducer(FlinkKafkaProducer011.java:904)
> =09at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.c=
reateOrGetProducerFromPool(FlinkKafkaProducer011.java:637)
> =09at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.b=
eginTransaction(FlinkKafkaProducer011.java:613)
> =09at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.b=
eginTransaction(FlinkKafkaProducer011.java:94)
> =09at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFun=
ction.beginTransactionInternal(TwoPhaseCommitSinkFunction.java:359)
> =09at org.apache.flink.streaming.api.functions.sink.TwoPhaseCommitSinkFun=
ction.snapshotState(TwoPhaseCommitSinkFunction.java:294)
> =09at org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer011.s=
napshotState(FlinkKafkaProducer011.java:756)
> =09at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.tr=
ySnapshotFunctionState(StreamingFunctionUtils.java:118)
> =09at org.apache.flink.streaming.util.functions.StreamingFunctionUtils.sn=
apshotFunctionState(StreamingFunctionUtils.java:99)
> =09at org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator.=
snapshotState(AbstractUdfStreamOperator.java:90)
> =09at org.apache.flink.streaming.api.operators.AbstractStreamOperator.sna=
pshotState(AbstractStreamOperator.java:357)
> =09... 12 more
> {noformat}
> *How to reproduce*
> Note that the problem only appears when a job is deployed on a cluster.=
=20
> # Build Flink 1.4
> # Build test job https://github.com/GJL/flink-kafka011-producer-test with=
 {{mvn -o clean install -Pbuild-jar}}
> # Start job:
> {noformat}
> bin/flink run -c com.garyyao.StreamingJob /pathto/flink-kafka011-producer=
/target/flink-kafka011-producer-1.0-SNAPSHOT.jar
> {noformat}


--
This message was sent by Atlassian JIRA
(v6.4.14#64029)