Mailing-List: contact user-help@ignite.apache.org; run by ezmlm
Precedence: bulk
Reply-To: user@ignite.apache.org
MIME-Version: 1.0
In-Reply-To: <CALPm_-CpHJE+Z=01g6FenAMUayX+HU3bDLn-tASkmftG=Kx6dA@mail.gmail.com>
References: <CALPm_-CpHJE+Z=01g6FenAMUayX+HU3bDLn-tASkmftG=Kx6dA@mail.gmail.com>
From: Anton Vinogradov <avinogradov@gridgain.com>
Date: Thu, 3 Nov 2016 13:29:51 +0300
Message-ID: <CAOKDqC-oiFWEf=Sm59myu1i8ngFB0UfNvr53kb5UtR5Qy3Q63A@mail.gmail.com>
Subject: Re: DataStreamer is closed
To: user <user@ignite.apache.org>
Content-Type: multipart/alternative; boundary=001a113fbc44b08612054063097e
archived-at: Thu, 03 Nov 2016 10:29:59 -0000

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

Anil,

Could you provide getStreamer() code and full logs?
Possible, ignite node was disconnected and this cause DataStreamer closure.

On Thu, Nov 3, 2016 at 1:17 PM, Anil <anilklce@gmail.com> wrote:

> HI,
>
> I have created custom kafka data streamer for my use case and i see
> following exception.
>
> java.lang.IllegalStateException: Data streamer has been closed.
>         at org.apache.ignite.internal.processors.datastreamer.
> DataStreamerImpl.enterBusy(DataStreamerImpl.java:360)
>         at org.apache.ignite.internal.processors.datastreamer.
> DataStreamerImpl.addData(DataStreamerImpl.java:507)
>         at org.apache.ignite.internal.processors.datastreamer.
> DataStreamerImpl.addData(DataStreamerImpl.java:498)
>         at net.juniper.cs.cache.KafkaCacheDataStreamer.addMessage(
> KafkaCacheDataStreamer.java:128)
>         at net.juniper.cs.cache.KafkaCacheDataStreamer$1.run(
> KafkaCacheDataStreamer.java:176)
>         at java.util.concurrent.Executors$RunnableAdapter.
> call(Executors.java:511)
>         at java.util.concurrent.FutureTask.run(FutureTask.java:266)
>         at java.util.concurrent.ThreadPoolExecutor.runWorker(
> ThreadPoolExecutor.java:1142)
>         at java.util.concurrent.ThreadPoolExecutor$Worker.run(
> ThreadPoolExecutor.java:617)
>         at java.lang.Thread.run(Thread.java:745)
>
>
>
> addMessage method is
>
>  @Override
>     protected void addMessage(T msg) {
>     if (getMultipleTupleExtractor() == null){
>             Map.Entry<K, V> e = getSingleTupleExtractor().extract(msg);
>
>             if (e != null)
>                 getStreamer().addData(e);
>
>         } else {
>             Map<K, V> m = getMultipleTupleExtractor().extract(msg);
>             if (m != null && !m.isEmpty()){
>                 getStreamer().addData(m);
>             }
>         }
>     }
>
>
> Do you see any issue ? Please let me know if you need any additional
> information. thanks.
>
> Thanks.
>

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

<div dir=3D"ltr"><span name=3D"Anil" class=3D"gmail-gD" style=3D"font-size:=
12.8px">Anil<span style=3D"white-space:nowrap">,=C2=A0</span></span><br><di=
v><span name=3D"Anil" class=3D"gmail-gD" style=3D"font-size:12.8px"><span s=
tyle=3D"white-space:nowrap"><br></span></span></div><div><span name=3D"Anil=
" class=3D"gmail-gD" style=3D"font-size:12.8px"><span style=3D"white-space:=
nowrap">Could you provide=C2=A0</span></span><span style=3D"font-size:12.8p=
x">getStreamer() code and full logs?</span></div><div><span style=3D"font-s=
ize:12.8px">Possible, ignite node was disconnected and this cause DataStrea=
mer closure.</span></div></div><div class=3D"gmail_extra"><br><div class=3D=
"gmail_quote">On Thu, Nov 3, 2016 at 1:17 PM, Anil <span dir=3D"ltr">&lt;<a=
 href=3D"mailto:anilklce@gmail.com" target=3D"_blank">anilklce@gmail.com</a=
>&gt;</span> wrote:<br><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,<=
div><br></div><div>I have created custom kafka data streamer for my use cas=
e and i see following exception.</div><div><br></div><div><div>java.lang.<w=
br>IllegalStateException: Data streamer has been closed.</div><div>=C2=A0 =
=C2=A0 =C2=A0 =C2=A0 at org.apache.ignite.internal.<wbr>processors.datastre=
amer.<wbr>DataStreamerImpl.enterBusy(<wbr>DataStreamerImpl.java:360)</div><=
div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at org.apache.ignite.internal.<wbr>processo=
rs.datastreamer.<wbr>DataStreamerImpl.addData(<wbr>DataStreamerImpl.java:50=
7)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at org.apache.ignite.internal.<wbr=
>processors.datastreamer.<wbr>DataStreamerImpl.addData(<wbr>DataStreamerImp=
l.java:498)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at net.juniper.cs.cache.<=
wbr>KafkaCacheDataStreamer.<wbr>addMessage(<wbr>KafkaCacheDataStreamer.java=
:<wbr>128)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at net.juniper.cs.cache.<w=
br>KafkaCacheDataStreamer$1.run(<wbr>KafkaCacheDataStreamer.java:<wbr>176)<=
/div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.concurrent.<wbr>Executor=
s$RunnableAdapter.<wbr>call(Executors.java:511)</div><div>=C2=A0 =C2=A0 =C2=
=A0 =C2=A0 at java.util.concurrent.<wbr>FutureTask.run(FutureTask.<wbr>java=
:266)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.concurrent.<wbr>Th=
readPoolExecutor.runWorker(<wbr>ThreadPoolExecutor.java:1142)</div><div>=C2=
=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.concurrent.<wbr>ThreadPoolExecutor$Wo=
rker.run(<wbr>ThreadPoolExecutor.java:617)</div><div>=C2=A0 =C2=A0 =C2=A0 =
=C2=A0 at java.lang.Thread.run(Thread.<wbr>java:745)</div></div><div><br></=
div><div><br></div><div><br></div><div>addMessage method is =C2=A0</div><di=
v><br></div><div><div>=C2=A0@Override</div><div>=C2=A0 =C2=A0 protected voi=
d addMessage(T msg) {</div><div>=C2=A0 =C2=A0 <span class=3D"m_420799275340=
325761gmail-Apple-tab-span" style=3D"white-space:pre-wrap">	</span>if (getM=
ultipleTupleExtractor() =3D=3D null){</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0=
 =C2=A0 =C2=A0 Map.Entry&lt;K, V&gt; e =3D getSingleTupleExtractor().<wbr>e=
xtract(msg);</div><div><br></div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 if (e !=3D null)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0=
 =C2=A0 =C2=A0 getStreamer().addData(e);</div><div><br></div><div>=C2=A0 =
=C2=A0 =C2=A0 =C2=A0 } else {</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 Map&lt;K, V&gt; m =3D getMultipleTupleExtractor().<wbr>extract(msg);=
</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 if (m !=3D null &amp;&=
amp; !m.isEmpty()){</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=
=A0 =C2=A0 getStreamer().addData(m);</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =
=C2=A0 =C2=A0 }</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 }</div><div>=C2=A0 =
=C2=A0 }</div></div><div><br></div><div><br></div><div>Do you see any issue=
 ? Please let me know if you need any additional information. thanks.</div>=
<div><br></div><div>Thanks.</div></div>
</blockquote></div><br></div>

--001a113fbc44b08612054063097e--