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: <CAOKDqC-oiFWEf=Sm59myu1i8ngFB0UfNvr53kb5UtR5Qy3Q63A@mail.gmail.com>
References: <CALPm_-CpHJE+Z=01g6FenAMUayX+HU3bDLn-tASkmftG=Kx6dA@mail.gmail.com>
 <CAOKDqC-oiFWEf=Sm59myu1i8ngFB0UfNvr53kb5UtR5Qy3Q63A@mail.gmail.com>
From: Anil <anilklce@gmail.com>
Date: Thu, 3 Nov 2016 16:11:54 +0530
Message-ID: <CALPm_-D3mB5QbZVY+i=Kk6XyMMSoQUO=8Tqu6X_psstApzrymw@mail.gmail.com>
Subject: Re: DataStreamer is closed
To: user@ignite.apache.org
Content-Type: multipart/mixed; boundary=001a114c98b82b1dba0540633751
archived-at: Thu, 03 Nov 2016 10:43:17 -0000

--001a114c98b82b1dba0540633751
Content-Type: multipart/alternative; boundary=001a114c98b82b1db6054063374f

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

Hi Anton,
No. ignite nodes looks good.

I have attached my KafkaCacheDataStreamer class and following is the code
to listen to the kafka topic. IgniteCache is created using java
configuration.

I see cache size is zero after adding the entries to cache as well from
KafkaCacheDataStreamer. Not sure how to log whether the entries added to
cache or not.

KafkaCacheDataStreamer<String, String, Person> kafkaStreamer = new
KafkaCacheDataStreamer<String, String, Person>();

 Properites pros = new Properites() // kafka properties
 ConsumerConfig consumerConfig = new ConsumerConfig(props);

    try {
    IgniteDataStreamer<String, Person> stmr =
ignite.dataStreamer(CacheManager.PERSON_CACHE);
       // allow overwriting cache data
       stmr.allowOverwrite(true);

       kafkaStreamer.setIgnite(ignite);
       kafkaStreamer.setStreamer(stmr);

       // set the topic
       kafkaStreamer.setTopic(kafkaConfig.getString("topic", "TestTopic"));

       // set the number of threads to process Kafka streams
       kafkaStreamer.setThreads(1);

       // set Kafka consumer configurations
       kafkaStreamer.setConsumerConfig(consumerConfig);

       // set decoders
       kafkaStreamer.setKeyDecoder(new StringDecoder(new
VerifiableProperties()));
       kafkaStreamer.setValueDecoder(new StringDecoder(new
VerifiableProperties()));
       kafkaStreamer.setMultipleTupleExtractor(new
StreamMultipleTupleExtractor<String, String, Person>() {
    @Override
    public Map<String, Person> extract(String msg) {
    Map<String, Person> entries = new HashMap<>();
    try {
    KafkaMessage request = Json.decodeValue(msg, KafkaMessage.class);
    IgniteCache<String, Person> cache = CacheManager.getCache();

    if (CollectionUtils.isNotEmpty(request.getPersons())){
    String id = null;
    for (Person ib : request.getPersons()){
    if (StringUtils.isNotBlank(ib.getId())){
    id = ib.getId();
    if (null != ib.isDeleted() && Boolean.TRUE.equals(ib.isDeleted())){
    cache.remove(id);
    }else {
    // no need to store the id. so setting null.
    ib.setId(null);
    entries.put(id, ib);
    }
    }
    }
    }else {

    }
    }catch (Exception ex){
    logger.error("Error while updating the cache - {} {} " ,msg, ex);
    }

    return entries;
    }
    });

       kafkaStreamer.start();
    }catch (Exception ex){
    logger.error("Error in kafka data streamer ", ex);
    }


Please let me know if you see any issues. thanks.

On 3 November 2016 at 15:59, Anton Vinogradov <avinogradov@gridgain.com>
wrote:

> 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.DataStrea
>> merImpl.enterBusy(DataStreamerImpl.java:360)
>>         at org.apache.ignite.internal.processors.datastreamer.DataStrea
>> merImpl.addData(DataStreamerImpl.java:507)
>>         at org.apache.ignite.internal.processors.datastreamer.DataStrea
>> merImpl.addData(DataStreamerImpl.java:498)
>>         at net.juniper.cs.cache.KafkaCacheDataStreamer.addMessage(Kafka
>> CacheDataStreamer.java:128)
>>         at net.juniper.cs.cache.KafkaCacheDataStreamer$1.run(KafkaCache
>> DataStreamer.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(ThreadPool
>> Executor.java:1142)
>>         at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoo
>> lExecutor.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.
>>
>
>

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

<div dir=3D"ltr">Hi Anton,<div>No. ignite nodes looks good.</div><div><br><=
/div><div>I have attached my KafkaCacheDataStreamer class and following is =
the code to listen to the kafka topic. IgniteCache is created using java co=
nfiguration.</div><div><br></div><div>I see cache size is zero after adding=
 the entries to cache as well from KafkaCacheDataStreamer. Not sure how to =
log whether the entries added to cache or not.</div><div><br></div><div><di=
v>KafkaCacheDataStreamer&lt;String, String, Person&gt; kafkaStreamer =3D ne=
w KafkaCacheDataStreamer&lt;String, String, Person&gt;();</div><div><br></d=
iv><div>=C2=A0Properites pros =3D new Properites() // kafka properties</div=
><div>=C2=A0ConsumerConfig consumerConfig =3D new ConsumerConfig(props);</d=
iv><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-s=
pace:pre">		</span></div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-=
span" style=3D"white-space:pre">		</span>try {</div><div>=C2=A0 =C2=A0 <spa=
n class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">			</span>Ignite=
DataStreamer&lt;String, Person&gt; stmr =3D ignite.dataStreamer(CacheManage=
r.PERSON_CACHE);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-spa=
n" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0// allow overwriting ca=
che data</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=
=3D"white-space:pre">		</span> =C2=A0 =C2=A0stmr.allowOverwrite(true);</div=
><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-spa=
ce:pre">		</span> =C2=A0 =C2=A0</div><div>=C2=A0 =C2=A0 <span class=3D"gmai=
l-Apple-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0kafkaStr=
eamer.setIgnite(ignite);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple=
-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0kafkaStreamer.s=
etStreamer(stmr);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-sp=
an" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0</div><div>=C2=A0 =C2=
=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">		</span=
> =C2=A0 =C2=A0// set the topic</div><div>=C2=A0 =C2=A0 <span class=3D"gmai=
l-Apple-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0kafkaStr=
eamer.setTopic(kafkaConfig.getString(&quot;topic&quot;, &quot;TestTopic&quo=
t;));</div><div><br></div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab=
-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0// set the number o=
f threads to process Kafka streams</div><div>=C2=A0 =C2=A0 <span class=3D"g=
mail-Apple-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0kafka=
Streamer.setThreads(1);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-=
tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0</div><div>=C2=
=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">	=
	</span> =C2=A0 =C2=A0// set Kafka consumer configurations</div><div>=C2=A0=
 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">		</=
span> =C2=A0 =C2=A0kafkaStreamer.setConsumerConfig(consumerConfig);</div><d=
iv>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:=
pre">		</span> =C2=A0 =C2=A0</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-A=
pple-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=A0// set deco=
ders</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"=
white-space:pre">		</span> =C2=A0 =C2=A0kafkaStreamer.setKeyDecoder(new Str=
ingDecoder(new VerifiableProperties()));</div><div>=C2=A0 =C2=A0 <span clas=
s=3D"gmail-Apple-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =C2=
=A0kafkaStreamer.setValueDecoder(new StringDecoder(new VerifiableProperties=
()));</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D=
"white-space:pre">		</span> =C2=A0 =C2=A0kafkaStreamer.setMultipleTupleExtr=
actor(new StreamMultipleTupleExtractor&lt;String, String, Person&gt;() {</d=
iv><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-s=
pace:pre">				</span>@Override</div><div>=C2=A0 =C2=A0 <span class=3D"gmail=
-Apple-tab-span" style=3D"white-space:pre">				</span>public Map&lt;String,=
 Person&gt; extract(String msg) {</div><div>=C2=A0 =C2=A0 <span class=3D"gm=
ail-Apple-tab-span" style=3D"white-space:pre">					</span>Map&lt;String, Pe=
rson&gt; entries =3D new HashMap&lt;&gt;();</div><div>=C2=A0 =C2=A0 <span c=
lass=3D"gmail-Apple-tab-span" style=3D"white-space:pre">					</span>try {</=
div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-=
space:pre">						</span>KafkaMessage request =3D Json.decodeValue(msg, Kafk=
aMessage.class);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-spa=
n" style=3D"white-space:pre">						</span>IgniteCache&lt;String, Person&gt;=
 cache =3D CacheManager.getCache();</div><div>=C2=A0 =C2=A0 <span class=3D"=
gmail-Apple-tab-span" style=3D"white-space:pre">						</span></div><div>=C2=
=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">	=
					</span>if (CollectionUtils.isNotEmpty(request.getPersons())){</div><di=
v>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:p=
re">							</span>String id =3D null;</div><div>=C2=A0 =C2=A0 <span class=
=3D"gmail-Apple-tab-span" style=3D"white-space:pre">							</span>for (Pers=
on ib : request.getPersons()){ =C2=A0 =C2=A0<span class=3D"gmail-Apple-tab-=
span" style=3D"white-space:pre">								</span></div><div>=C2=A0 =C2=A0 <sp=
an class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">								</span>=
if (StringUtils.isNotBlank(ib.getId())){</div><div>=C2=A0 =C2=A0 <span clas=
s=3D"gmail-Apple-tab-span" style=3D"white-space:pre">									</span>id =3D=
 ib.getId();</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" s=
tyle=3D"white-space:pre">									</span>if (null !=3D ib.isDeleted() &amp;=
&amp; Boolean.TRUE.equals(ib.isDeleted())){</div><div>=C2=A0 =C2=A0 <span c=
lass=3D"gmail-Apple-tab-span" style=3D"white-space:pre">										</span>ca=
che.remove(id);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span=
" style=3D"white-space:pre">									</span>}else {</div><div>=C2=A0 =C2=A0=
 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">										<=
/span>// no need to store the id. so setting null.</div><div>=C2=A0 =C2=A0 =
<span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">										</=
span>ib.setId(null);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab=
-span" style=3D"white-space:pre">										</span>entries.put(id, ib);</div=
><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-spa=
ce:pre">									</span>}</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Appl=
e-tab-span" style=3D"white-space:pre">								</span>}</div><div>=C2=A0 =C2=
=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">							<=
/span>}</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=
=3D"white-space:pre">						</span>}else {</div><div>=C2=A0 =C2=A0 <span cla=
ss=3D"gmail-Apple-tab-span" style=3D"white-space:pre">							</span></div><=
div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space=
:pre">						</span>}<span class=3D"gmail-Apple-tab-span" style=3D"white-spa=
ce:pre">	</span></div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-spa=
n" style=3D"white-space:pre">					</span>}catch (Exception ex){</div><div>=
=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre=
">						</span>logger.error(&quot;Error while updating the cache - {} {} &q=
uot; ,msg, ex);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span=
" style=3D"white-space:pre">					</span>}</div><div>=C2=A0 =C2=A0 <span cla=
ss=3D"gmail-Apple-tab-span" style=3D"white-space:pre">					</span></div><di=
v>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:p=
re">					</span>return entries;</div><div>=C2=A0 =C2=A0 <span class=3D"gmai=
l-Apple-tab-span" style=3D"white-space:pre">				</span>}</div><div>=C2=A0 =
=C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">			</=
span>});</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=
=3D"white-space:pre">		</span> =C2=A0 =C2=A0</div><div>=C2=A0 =C2=A0 <span =
class=3D"gmail-Apple-tab-span" style=3D"white-space:pre">		</span> =C2=A0 =
=C2=A0kafkaStreamer.start();</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-A=
pple-tab-span" style=3D"white-space:pre">		</span>}catch (Exception ex){</d=
iv><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D"white-s=
pace:pre">			</span>logger.error(&quot;Error in kafka data streamer &quot;,=
 ex);</div><div>=C2=A0 =C2=A0 <span class=3D"gmail-Apple-tab-span" style=3D=
"white-space:pre">		</span>}</div></div><div><br></div><div><br></div><div>=
Please let me know if you see any issues. thanks.</div></div><div class=3D"=
gmail_extra"><br><div class=3D"gmail_quote">On 3 November 2016 at 15:59, An=
ton Vinogradov <span dir=3D"ltr">&lt;<a href=3D"mailto:avinogradov@gridgain=
.com" target=3D"_blank">avinogradov@gridgain.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"><span name=3D"Anil" class=3D=
"m_-8139851037813737596gmail-gD" style=3D"font-size:12.8px">Anil<span style=
=3D"white-space:nowrap">,=C2=A0</span></span><br><div><span name=3D"Anil" c=
lass=3D"m_-8139851037813737596gmail-gD" style=3D"font-size:12.8px"><span st=
yle=3D"white-space:nowrap"><br></span></span></div><div><span name=3D"Anil"=
 class=3D"m_-8139851037813737596gmail-gD" style=3D"font-size:12.8px"><span =
style=3D"white-space:nowrap">Could you provide=C2=A0</span></span><span sty=
le=3D"font-size:12.8px">getStreamer() code and full logs?</span></div><div>=
<span style=3D"font-size:12.8px">Possible, ignite node was disconnected and=
 this cause DataStreamer closure.</span></div></div><div class=3D"gmail_ext=
ra"><br><div class=3D"gmail_quote">On Thu, Nov 3, 2016 at 1:17 PM, Anil <sp=
an 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_quot=
e" 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 st=
reamer for my use case and i see following exception.</div><div><br></div><=
div><div>java.lang.IllegalStateExceptio<wbr>n: Data streamer has been close=
d.</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at <a href=3D"http://org.apache.ig=
nite.internal.pro">org.apache.ignite.internal.pro</a><wbr>cessors.datastrea=
mer.DataStrea<wbr>merImpl.enterBusy(DataStreamer<wbr>Impl.java:360)</div><d=
iv>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at <a href=3D"http://org.apache.ignite.inter=
nal.pro">org.apache.ignite.internal.pro</a><wbr>cessors.datastreamer.DataSt=
rea<wbr>merImpl.addData(DataStreamerIm<wbr>pl.java:507)</div><div>=C2=A0 =
=C2=A0 =C2=A0 =C2=A0 at <a href=3D"http://org.apache.ignite.internal.pro">o=
rg.apache.ignite.internal.pro</a><wbr>cessors.datastreamer.DataStrea<wbr>me=
rImpl.addData(DataStreamerIm<wbr>pl.java:498)</div><div>=C2=A0 =C2=A0 =C2=
=A0 =C2=A0 at net.juniper.cs.cache.KafkaCach<wbr>eDataStreamer.addMessage(K=
afka<wbr>CacheDataStreamer.java:128)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =
at net.juniper.cs.cache.KafkaCach<wbr>eDataStreamer$1.run(KafkaCache<wbr>Da=
taStreamer.java:176)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.con=
current.Executors<wbr>$RunnableAdapter.call(<wbr>Executors.java:511)</div><=
div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.concurrent.FutureTas<wbr>k.run=
(FutureTask.java:266)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.co=
ncurrent.ThreadPoo<wbr>lExecutor.runWorker(ThreadPool<wbr>Executor.java:114=
2)</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 at java.util.concurrent.ThreadPoo<=
wbr>lExecutor$Worker.run(ThreadPoo<wbr>lExecutor.java:617)</div><div>=C2=A0=
 =C2=A0 =C2=A0 =C2=A0 at java.lang.Thread.run(Thread.ja<wbr>va:745)</div></=
div><div><br></div><div><br></div><div><br></div><div>addMessage method is =
=C2=A0</div><div><br></div><div><div>=C2=A0@Override</div><div>=C2=A0 =C2=
=A0 protected void addMessage(T msg) {</div><div>=C2=A0 =C2=A0 <span class=
=3D"m_-8139851037813737596m_420799275340325761gmail-Apple-tab-span" style=
=3D"white-space:pre-wrap">	</span>if (getMultipleTupleExtractor() =3D=3D nu=
ll){</div><div>=C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 =C2=A0 Map.Entry&lt;K, V&=
gt; e =3D getSingleTupleExtractor().extr<wbr>act(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().addDa=
ta(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 getMult=
ipleTupleExtractor().ex<wbr>tract(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 a=
ny additional information. thanks.</div><div><br></div><div>Thanks.</div></=
div>
</blockquote></div><br></div>
</blockquote></div><br></div>

--001a114c98b82b1db6054063374f--

--001a114c98b82b1dba0540633751
Content-Type: application/octet-stream; name="KafkaDataSteamer.java"
Content-Disposition: attachment; filename="KafkaDataSteamer.java"
Content-Transfer-Encoding: base64
X-Attachment-Id: f_iv282aiz0

DQppbXBvcnQgamF2YS51dGlsLkhhc2hNYXA7DQppbXBvcnQgamF2YS51dGlsLkxpc3Q7DQppbXBv
cnQgamF2YS51dGlsLk1hcDsNCmltcG9ydCBqYXZhLnV0aWwuT2JqZWN0czsNCmltcG9ydCBqYXZh
LnV0aWwuY29uY3VycmVudC5FeGVjdXRvclNlcnZpY2U7DQppbXBvcnQgamF2YS51dGlsLmNvbmN1
cnJlbnQuRXhlY3V0b3JzOw0KaW1wb3J0IGphdmEudXRpbC5jb25jdXJyZW50LlRpbWVVbml0Ow0K
DQppbXBvcnQgb3JnLmFwYWNoZS5pZ25pdGUuSWduaXRlRXhjZXB0aW9uOw0KaW1wb3J0IG9yZy5h
cGFjaGUuaWduaXRlLnN0cmVhbS5TdHJlYW1BZGFwdGVyOw0KaW1wb3J0IG9yZy5zbGY0ai5Mb2dn
ZXI7DQppbXBvcnQgb3JnLnNsZjRqLkxvZ2dlckZhY3Rvcnk7DQoNCmltcG9ydCBrYWZrYS5jb25z
dW1lci5Db25zdW1lckNvbmZpZzsNCmltcG9ydCBrYWZrYS5jb25zdW1lci5Db25zdW1lckl0ZXJh
dG9yOw0KaW1wb3J0IGthZmthLmNvbnN1bWVyLkthZmthU3RyZWFtOw0KaW1wb3J0IGthZmthLmph
dmFhcGkuY29uc3VtZXIuQ29uc3VtZXJDb25uZWN0b3I7DQppbXBvcnQga2Fma2EubWVzc2FnZS5N
ZXNzYWdlQW5kTWV0YWRhdGE7DQppbXBvcnQga2Fma2Euc2VyaWFsaXplci5EZWNvZGVyOw0KDQpw
dWJsaWMgY2xhc3MgS2Fma2FDYWNoZURhdGFTdHJlYW1lcjxULCBLLCBWPiBleHRlbmRzIFN0cmVh
bUFkYXB0ZXI8VCwgSywgVj4gew0KICAgIA0KCXByaXZhdGUgc3RhdGljIExvZ2dlciBsb2cgPSBM
b2dnZXJGYWN0b3J5LmdldExvZ2dlcihLYWZrYUNhY2hlRGF0YVN0cmVhbWVyLmNsYXNzKTsNCgkN
CgkvKiogUmV0cnkgdGltZW91dC4gKi8NCiAgICBwcml2YXRlIHN0YXRpYyBmaW5hbCBsb25nIERG
TFRfUkVUUllfVElNRU9VVCA9IDEwMDAwOw0KDQogICAgLyoqIEV4ZWN1dG9yIHVzZWQgdG8gc3Vi
bWl0IGthZmthIHN0cmVhbXMuICovDQogICAgcHJpdmF0ZSBFeGVjdXRvclNlcnZpY2UgZXhlY3V0
b3I7DQoNCiAgICAvKiogVG9waWMuICovDQogICAgcHJpdmF0ZSBTdHJpbmcgdG9waWM7DQoNCiAg
ICAvKiogTnVtYmVyIG9mIHRocmVhZHMgdG8gcHJvY2VzcyBrYWZrYSBzdHJlYW1zLiAqLw0KICAg
IHByaXZhdGUgaW50IHRocmVhZHM7DQoNCiAgICAvKiogS2Fma2EgY29uc3VtZXIgY29uZmlnLiAq
Lw0KICAgIHByaXZhdGUgQ29uc3VtZXJDb25maWcgY29uc3VtZXJDZmc7DQoNCiAgICAvKiogS2V5
IGRlY29kZXIuICovDQogICAgcHJpdmF0ZSBEZWNvZGVyPFQ+IGtleURlY29kZXI7DQoNCiAgICAv
KiogVmFsdWUgZGVjb2Rlci4gKi8NCiAgICBwcml2YXRlIERlY29kZXI8VD4gdmFsRGVjb2RlcjsN
Cg0KICAgIC8qKiBLYWZrYSBjb25zdW1lciBjb25uZWN0b3IuICovDQogICAgcHJpdmF0ZSBDb25z
dW1lckNvbm5lY3RvciBjb25zdW1lcjsNCg0KICAgIC8qKiBSZXRyeSB0aW1lb3V0LiAqLw0KICAg
IHByaXZhdGUgbG9uZyByZXRyeVRpbWVvdXQgPSBERkxUX1JFVFJZX1RJTUVPVVQ7DQoNCiAgICAv
KiogU3RvcHBlZC4gKi8NCiAgICBwcml2YXRlIHZvbGF0aWxlIGJvb2xlYW4gc3RvcHBlZDsNCg0K
ICAgIC8qKg0KICAgICAqIFNldHMgdGhlIHRvcGljIG5hbWUuDQogICAgICoNCiAgICAgKiBAcGFy
YW0gdG9waWMgVG9waWMgbmFtZS4NCiAgICAgKi8NCiAgICBwdWJsaWMgdm9pZCBzZXRUb3BpYyhT
dHJpbmcgdG9waWMpIHsNCiAgICAgICAgdGhpcy50b3BpYyA9IHRvcGljOw0KICAgIH0NCg0KICAg
IC8qKg0KICAgICAqIFNldHMgdGhlIHRocmVhZHMuDQogICAgICoNCiAgICAgKiBAcGFyYW0gdGhy
ZWFkcyBOdW1iZXIgb2YgdGhyZWFkcy4NCiAgICAgKi8NCiAgICBwdWJsaWMgdm9pZCBzZXRUaHJl
YWRzKGludCB0aHJlYWRzKSB7DQogICAgICAgIHRoaXMudGhyZWFkcyA9IHRocmVhZHM7DQogICAg
fQ0KDQogICAgLyoqDQogICAgICogU2V0cyB0aGUgY29uc3VtZXIgY29uZmlnLg0KICAgICAqDQog
ICAgICogQHBhcmFtIGNvbnN1bWVyQ2ZnIENvbnN1bWVyIGNvbmZpZ3VyYXRpb24uDQogICAgICov
DQogICAgcHVibGljIHZvaWQgc2V0Q29uc3VtZXJDb25maWcoQ29uc3VtZXJDb25maWcgY29uc3Vt
ZXJDZmcpIHsNCiAgICAgICAgdGhpcy5jb25zdW1lckNmZyA9IGNvbnN1bWVyQ2ZnOw0KICAgIH0N
Cg0KICAgIC8qKg0KICAgICAqIFNldHMgdGhlIGtleSBkZWNvZGVyLg0KICAgICAqDQogICAgICog
QHBhcmFtIGtleURlY29kZXIgS2V5IGRlY29kZXIuDQogICAgICovDQogICAgcHVibGljIHZvaWQg
c2V0S2V5RGVjb2RlcihEZWNvZGVyPFQ+IGtleURlY29kZXIpIHsNCiAgICAgICAgdGhpcy5rZXlE
ZWNvZGVyID0ga2V5RGVjb2RlcjsNCiAgICB9DQoNCiAgICAvKioNCiAgICAgKiBTZXRzIHRoZSB2
YWx1ZSBkZWNvZGVyLg0KICAgICAqDQogICAgICogQHBhcmFtIHZhbERlY29kZXIgVmFsdWUgZGVj
b2Rlci4NCiAgICAgKi8NCiAgICBwdWJsaWMgdm9pZCBzZXRWYWx1ZURlY29kZXIoRGVjb2RlcjxU
PiB2YWxEZWNvZGVyKSB7DQogICAgICAgIHRoaXMudmFsRGVjb2RlciA9IHZhbERlY29kZXI7DQog
ICAgfQ0KDQogICAgLyoqDQogICAgICogU2V0cyB0aGUgcmV0cnkgdGltZW91dC4NCiAgICAgKg0K
ICAgICAqIEBwYXJhbSByZXRyeVRpbWVvdXQgUmV0cnkgdGltZW91dC4NCiAgICAgKi8NCiAgICBw
dWJsaWMgdm9pZCBzZXRSZXRyeVRpbWVvdXQobG9uZyByZXRyeVRpbWVvdXQpIHsNCi8vICAgICAg
ICBBLmVuc3VyZShyZXRyeVRpbWVvdXQgPiAwLCAicmV0cnlUaW1lb3V0ID4gMCIpOw0KICAgIAlp
ZiAocmV0cnlUaW1lb3V0IDw9IDApew0KICAgIAkJdGhyb3cgbmV3IElsbGVnYWxBcmd1bWVudEV4
Y2VwdGlvbigicmV0cnlUaW1lb3V0IG1pc3QgYmUgZ3JlYXRlciB0aGFuIDAiKTsNCiAgICAJCQ0K
ICAgIAl9DQogICAgICAgIHRoaXMucmV0cnlUaW1lb3V0ID0gcmV0cnlUaW1lb3V0Ow0KICAgIH0N
Cg0KICAgIA0KICAgIEBPdmVycmlkZQ0KICAgIHByb3RlY3RlZCB2b2lkIGFkZE1lc3NhZ2UoVCBt
c2cpIHsNCiAgICAJdHJ5IHsNCiAgICAJCWlmIChnZXRNdWx0aXBsZVR1cGxlRXh0cmFjdG9yKCkg
PT0gbnVsbCl7DQogICAgICAgICAgICAgICAgTWFwLkVudHJ5PEssIFY+IGUgPSBnZXRTaW5nbGVU
dXBsZUV4dHJhY3RvcigpLmV4dHJhY3QobXNnKTsNCg0KICAgICAgICAgICAgICAgIGlmIChlICE9
IG51bGwpDQogICAgICAgICAgICAgICAgICAgIGdldFN0cmVhbWVyKCkuYWRkRGF0YShlKTsNCg0K
ICAgICAgICAgICAgfSBlbHNlIHsNCiAgICAgICAgICAgICAgICBNYXA8SywgVj4gbSA9IGdldE11
bHRpcGxlVHVwbGVFeHRyYWN0b3IoKS5leHRyYWN0KG1zZyk7DQogICAgICAgICAgICAgICAgaWYg
KG0gIT0gbnVsbCAmJiAhbS5pc0VtcHR5KCkpew0KICAgICAgICAgICAgICAgICAgICBnZXRTdHJl
YW1lcigpLmFkZERhdGEobSk7DQogICAgICAgICAgICAgICAgfQ0KICAgICAgICAgICAgfQ0KCQ0K
ICAgIAl9Y2F0Y2goRXhjZXB0aW9uIGV4KXsNCiAgICAJCWxvZy5lcnJvcigiRXhjZXB0aW9uIHdo
aWxlIGFkZGluZyB0byBzdHJlYW1lciAiLCBleCk7DQogICAgCX0NCiAgICB9DQogICAgLyoqDQog
ICAgICogU3RhcnRzIHN0cmVhbWVyLg0KICAgICAqDQogICAgICogQHRocm93cyBJZ25pdGVFeGNl
cHRpb24gSWYgZmFpbGVkLg0KICAgICAqLw0KICAgIHB1YmxpYyB2b2lkIHN0YXJ0KCkgew0KICAg
ICAgICBPYmplY3RzLnJlcXVpcmVOb25OdWxsKGdldFN0cmVhbWVyKCksICJzdHJlYW1lciIpOw0K
ICAgICAgICBPYmplY3RzLnJlcXVpcmVOb25OdWxsKGdldElnbml0ZSgpLCAiaWduaXRlIik7DQog
ICAgICAgIE9iamVjdHMucmVxdWlyZU5vbk51bGwodG9waWMsICJ0b3BpYyIpOw0KICAgICAgICBP
YmplY3RzLnJlcXVpcmVOb25OdWxsKGtleURlY29kZXIsICJrZXkgZGVjb2RlciIpOw0KICAgICAg
ICBPYmplY3RzLnJlcXVpcmVOb25OdWxsKHZhbERlY29kZXIsICJ2YWx1ZSBkZWNvZGVyIik7DQog
ICAgICAgIA0KICAgICAgICBPYmplY3RzLnJlcXVpcmVOb25OdWxsKGNvbnN1bWVyQ2ZnLCAia2Fm
a2EgY29uc3VtZXIgY29uZmlnIik7DQogICAgICAgIGlmKHRocmVhZHMgPD0gMCl7DQogICAgICAg
IAl0aHJvdyBuZXcgSWxsZWdhbEFyZ3VtZW50RXhjZXB0aW9uKCJUaHJlYWRzIGNhbm5vdCBiZSB6
ZXJvIik7DQogICAgICAgIH0NCg0KICAgICAgICBjb25zdW1lciA9IGthZmthLmNvbnN1bWVyLkNv
bnN1bWVyLmNyZWF0ZUphdmFDb25zdW1lckNvbm5lY3Rvcihjb25zdW1lckNmZyk7DQoNCiAgICAg
ICAgTWFwPFN0cmluZywgSW50ZWdlcj4gdG9waWNDbnRNYXAgPSBuZXcgSGFzaE1hcDw+KCk7DQoN
CiAgICAgICAgdG9waWNDbnRNYXAucHV0KHRvcGljLCB0aHJlYWRzKTsNCg0KICAgICAgICBNYXA8
U3RyaW5nLCBMaXN0PEthZmthU3RyZWFtPFQsIFQ+Pj4gY29uc3VtZXJNYXAgPQ0KICAgICAgICAg
ICAgY29uc3VtZXIuY3JlYXRlTWVzc2FnZVN0cmVhbXModG9waWNDbnRNYXAsIGtleURlY29kZXIs
IHZhbERlY29kZXIpOw0KDQogICAgICAgIExpc3Q8S2Fma2FTdHJlYW08VCwgVD4+IHN0cmVhbXMg
PSBjb25zdW1lck1hcC5nZXQodG9waWMpOw0KICAgICAgICANCiAgICAgICAgbG9nLmluZm8oIkth
ZmthIGlzIGNvbm5lY3RlZCBzdWNjZXNzZnVsbHkiKTsgICAgICAgICAgICAgICAgDQogICAgICAg
IGV4ZWN1dG9yID0gRXhlY3V0b3JzLm5ld0ZpeGVkVGhyZWFkUG9vbCh0aHJlYWRzKTsNCg0KICAg
ICAgICBzdG9wcGVkID0gZmFsc2U7DQoNCiAgICAgICAgLy8gTm93IGNyZWF0ZSBhbiBvYmplY3Qg
dG8gY29uc3VtZSB0aGUgbWVzc2FnZXMuDQogICAgICAgIGZvciAoZmluYWwgS2Fma2FTdHJlYW08
VCwgVD4gc3RyZWFtIDogc3RyZWFtcykgew0KICAgICAgICAgICAgZXhlY3V0b3Iuc3VibWl0KG5l
dyBSdW5uYWJsZSgpIHsNCiAgICAgICAgICAgICAgICBAT3ZlcnJpZGUgcHVibGljIHZvaWQgcnVu
KCkgew0KICAgICAgICAgICAgICAgICAgICB3aGlsZSAoIXN0b3BwZWQpIHsNCiAgICAgICAgICAg
ICAgICAgICAgICAgIHRyeSB7DQogICAgICAgICAgICAgICAgICAgICAgICAgICAgZm9yIChDb25z
dW1lckl0ZXJhdG9yPFQsIFQ+IGl0ID0gc3RyZWFtLml0ZXJhdG9yKCk7IGl0Lmhhc05leHQoKSAm
JiAhc3RvcHBlZDsgKSB7DQogICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIE1lc3NhZ2VB
bmRNZXRhZGF0YTxULCBUPiBtc2cgPSBpdC5uZXh0KCk7DQogICAgICAgICAgICAgICAgICAgICAg
ICAgICAgICAgIGxvZy5pbmZvKCJNZXNzYWdlIHJlY2lldmVkIHt9ICIsIG1zZy5tZXNzYWdlKCkp
OyANCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgdHJ5IHsNCiAgICAgICAgICAgICAg
ICAgICAgICAgICAgICAgICAgCWFkZE1lc3NhZ2UobXNnLm1lc3NhZ2UoKSk7DQovLyAgICAgICAg
ICAgICAgICAgICAgICAgICAgICAgICAgICAgIGdldFN0cmVhbWVyKCkuYWRkRGF0YShtc2cua2V5
KCksIG1zZy5tZXNzYWdlKCkpOw0KICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICB9DQog
ICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgIGNhdGNoIChFeGNlcHRpb24gZSkgew0KICAg
ICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgbG9nLmVycm9yKCJNZXNzYWdlIGlzIGln
bm9yZWQgZHVlIHRvIGFuIGVycm9yIFttc2c9IiArIG1zZyArICddJywgZSk7DQogICAgICAgICAg
ICAgICAgICAgICAgICAgICAgICAgIH0NCiAgICAgICAgICAgICAgICAgICAgICAgICAgICB9DQog
ICAgICAgICAgICAgICAgICAgICAgICB9DQogICAgICAgICAgICAgICAgICAgICAgICBjYXRjaCAo
RXhjZXB0aW9uIGUpIHsNCiAgICAgICAgICAgICAgICAgICAgICAgICAgICBsb2cuZXJyb3IoIk1l
c3NhZ2UgY2FuJ3QgYmUgY29uc3VtZWQgZnJvbSBzdHJlYW0uIFJldHJ5IGFmdGVyICIgKyByZXRy
eVRpbWVvdXQgKyAiIG1zLiIsIGUpOw0KDQogICAgICAgICAgICAgICAgICAgICAgICAgICAgdHJ5
IHsNCiAgICAgICAgICAgICAgICAgICAgICAgICAgICAgICAgVGhyZWFkLnNsZWVwKHJldHJ5VGlt
ZW91dCk7DQogICAgICAgICAgICAgICAgICAgICAgICAgICAgfQ0KICAgICAgICAgICAgICAgICAg
ICAgICAgICAgIGNhdGNoIChJbnRlcnJ1cHRlZEV4Y2VwdGlvbiBpZSkgew0KICAgICAgICAgICAg
ICAgICAgICAgICAgICAgICAgICAvLyBOby1vcC4NCiAgICAgICAgICAgICAgICAgICAgICAgICAg
ICB9DQogICAgICAgICAgICAgICAgICAgICAgICB9DQogICAgICAgICAgICAgICAgICAgIH0NCiAg
ICAgICAgICAgICAgICB9DQogICAgICAgICAgICB9KTsNCiAgICAgICAgfQ0KICAgIH0NCg0KICAg
IC8qKg0KICAgICAqIFN0b3BzIHN0cmVhbWVyLg0KICAgICAqLw0KICAgIHB1YmxpYyB2b2lkIHN0
b3AoKSB7DQogICAgICAgIHN0b3BwZWQgPSB0cnVlOw0KDQogICAgICAgIGlmIChjb25zdW1lciAh
PSBudWxsKQ0KICAgICAgICAgICAgY29uc3VtZXIuc2h1dGRvd24oKTsNCg0KICAgICAgICBpZiAo
ZXhlY3V0b3IgIT0gbnVsbCkgew0KICAgICAgICAgICAgZXhlY3V0b3Iuc2h1dGRvd24oKTsNCg0K
ICAgICAgICAgICAgdHJ5IHsNCiAgICAgICAgICAgICAgICBpZiAoIWV4ZWN1dG9yLmF3YWl0VGVy
bWluYXRpb24oNTAwMCwgVGltZVVuaXQuTUlMTElTRUNPTkRTKSkNCiAgICAgICAgICAgICAgICAg
ICAgaWYgKGxvZy5pc0RlYnVnRW5hYmxlZCgpKQ0KICAgICAgICAgICAgICAgICAgICAgICAgbG9n
LmRlYnVnKCJUaW1lZCBvdXQgd2FpdGluZyBmb3IgY29uc3VtZXIgdGhyZWFkcyB0byBzaHV0IGRv
d24sIGV4aXRpbmcgdW5jbGVhbmx5LiIpOw0KICAgICAgICAgICAgfQ0KICAgICAgICAgICAgY2F0
Y2ggKEludGVycnVwdGVkRXhjZXB0aW9uIGUpIHsNCiAgICAgICAgICAgICAgICBpZiAobG9nLmlz
RGVidWdFbmFibGVkKCkpDQogICAgICAgICAgICAgICAgICAgIGxvZy5kZWJ1ZygiSW50ZXJydXB0
ZWQgZHVyaW5nIHNodXRkb3duLCBleGl0aW5nIHVuY2xlYW5seS4iKTsNCiAgICAgICAgICAgIH0N
CiAgICAgICAgfQ0KICAgIH0NCg0KfQ0K
--001a114c98b82b1dba0540633751--