Mailing-List: contact user-help@storm.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@storm.apache.org
MIME-Version: 1.0
References: <CAP19eqyAMpVG1SQ3xzJ-eCYKcADp8u0pv24gbGwHfVcL9aWoXw@mail.gmail.com>
In-Reply-To: <CAP19eqyAMpVG1SQ3xzJ-eCYKcADp8u0pv24gbGwHfVcL9aWoXw@mail.gmail.com>
From: =?UTF-8?Q?Aaron_Niskod=C3=A9=2DDossett?= <dossett@gmail.com>
Date: Tue, 06 Sep 2016 13:39:56 +0000
Message-ID: <CA+s9jkSyVhmy_PVo4iqqXDgr6HE+nZ24-Bo2OiciPEOnV4Vr0g@mail.gmail.com>
Subject: Re: Increasing worker parallelism decreases throughput and increases
 tuple timeout
To: user@storm.apache.org
Content-Type: multipart/alternative; boundary=001a113d525a4869c2053bd6ef02
archived-at: Tue, 06 Sep 2016 13:40:20 -0000

--001a113d525a4869c2053bd6ef02
Content-Type: text/plain; charset=UTF-8

Hi Kris,

One possibility is that the Serializer isn't actually caching the schema
<-> id mappings and is hitting the schema registry every time.  The call to
register() in getFingerprint() [1] in particular can be a finicky since the
cache is ultimately in an IDENTITY hash map, not a regular old hashmap[2].
I'm familiar with the Avro deserializer you're using and though it
accounted for this, but perhaps not.

You could add timing information to the getFingerprint() and getSchema()
calls in ConfluentAvroSerializer.  If the results indicate cache misses,
that's probably your culprit.

Best, Aaron


[1]
https://github.com/apache/storm/blob/master/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/avro/ConfluentAvroSerializer.java#L66
[2]
https://github.com/confluentinc/schema-registry/blob/v1.0/client/src/main/java/io/confluent/kafka/schemaregistry/client/CachedSchemaRegistryClient.java#L79

On Tue, Sep 6, 2016 at 7:40 AM Kristopher Kane <kkane.list@gmail.com> wrote:

> Hi everyone.
>
> I have a simple topology that uses the Avro serializer (
> https://github.com/apache/storm/blob/master/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/avro/ConfluentAvroSerializer.java)
> and writes to Elasticsearch.
>
> The topology is like this:
>
> Kafka (raw scheme) -> Avro deserializer -> Elasticsearch
>
> This topology runs well with one worker, however, once I add one more
> worker (total of two) and change nothing else, the topology throughput
> drops and tuples start timing out.
>
> I've attached visualvm/jstatd to the workers when in multi worker mode -
> and added some jmx configs to the worker opts - but I am unable to see
> anything glaring.
>
> I've never seen Storm act this way but have also never worked with a
> custom serializer so assume that it is the culprit but I cannot explain
> why.
>
> Any pointers would be appreciated.
>
> Kris
>
>
>
>
>

--001a113d525a4869c2053bd6ef02
Content-Type: text/html; charset=UTF-8
Content-Transfer-Encoding: quoted-printable

<div dir=3D"ltr">Hi Kris,<div><br></div><div>One possibility is that the Se=
rializer isn&#39;t actually caching the schema &lt;-&gt; id mappings and is=
 hitting the schema registry every time.=C2=A0 The call to register() in ge=
tFingerprint() [1] in particular can be a finicky since the cache is ultima=
tely in an IDENTITY hash map, not a regular old hashmap[2].=C2=A0 I&#39;m f=
amiliar with the Avro deserializer you&#39;re using and though it accounted=
 for this, but perhaps not.</div><div><br></div><div>You could add timing i=
nformation to the getFingerprint() and getSchema() calls in ConfluentAvroSe=
rializer.=C2=A0 If the results indicate cache misses, that&#39;s probably y=
our culprit.</div><div><br></div><div>Best, Aaron</div><div><br></div><div>=
<br></div><div>[1] <a href=3D"https://github.com/apache/storm/blob/master/e=
xternal/storm-hdfs/src/main/java/org/apache/storm/hdfs/avro/ConfluentAvroSe=
rializer.java#L66">https://github.com/apache/storm/blob/master/external/sto=
rm-hdfs/src/main/java/org/apache/storm/hdfs/avro/ConfluentAvroSerializer.ja=
va#L66</a><br></div><div>[2]=C2=A0<a href=3D"https://github.com/confluentin=
c/schema-registry/blob/v1.0/client/src/main/java/io/confluent/kafka/schemar=
egistry/client/CachedSchemaRegistryClient.java#L79">https://github.com/conf=
luentinc/schema-registry/blob/v1.0/client/src/main/java/io/confluent/kafka/=
schemaregistry/client/CachedSchemaRegistryClient.java#L79</a></div></div><b=
r><div class=3D"gmail_quote"><div dir=3D"ltr">On Tue, Sep 6, 2016 at 7:40 A=
M Kristopher Kane &lt;<a href=3D"mailto:kkane.list@gmail.com">kkane.list@gm=
ail.com</a>&gt; wrote:<br></div><blockquote class=3D"gmail_quote" style=3D"=
margin:0 0 0 .8ex;border-left:1px #ccc solid;padding-left:1ex"><div dir=3D"=
ltr">Hi everyone. =C2=A0<div><br></div><div>I have a simple topology that u=
ses the Avro serializer (<a href=3D"https://github.com/apache/storm/blob/ma=
ster/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/avro/Confluent=
AvroSerializer.java" target=3D"_blank">https://github.com/apache/storm/blob=
/master/external/storm-hdfs/src/main/java/org/apache/storm/hdfs/avro/Conflu=
entAvroSerializer.java</a>)=C2=A0 and writes to Elasticsearch.=C2=A0</div><=
div><br></div><div>The topology is like this:</div><div><br></div><div>Kafk=
a (raw scheme) -&gt; Avro deserializer -&gt; Elasticsearch</div><div><br></=
div><div>This topology runs well with one worker, however, once I add one m=
ore worker (total of two) and change nothing else, the topology throughput =
drops and tuples start timing out. =C2=A0</div><div><br></div><div>I&#39;ve=
 attached visualvm/jstatd to the workers when in multi worker mode - and ad=
ded some jmx configs to the worker opts - but I am unable to see anything g=
laring.</div><div><br></div><div>I&#39;ve never seen Storm act this way but=
 have also never worked with a custom serializer so assume that it is the c=
ulprit but I cannot explain why.=C2=A0</div><div><br></div><div>Any pointer=
s would be appreciated.=C2=A0</div><div><br></div><div>Kris</div><div><br><=
/div><div><br></div><div><br></div><div><br></div></div>
</blockquote></div>

--001a113d525a4869c2053bd6ef02--