You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Sameer Kumar <sa...@gmail.com> on 2017/08/14 05:24:12 UTC
Synchronized methods in RockSB key store
Hi,
I was inspecting RocksDBStore and I saw get and put methods being
synchronized.
Are there any issues in RocksDB due to which we have used it. Could someone
please enlighten more on this.
@Override
public synchronized V get(K key) {
validateStoreOpen();
byte[] byteValue = getInternal(serdes.rawKey(key));
if (byteValue == null) {
return null;
} else {
return serdes.valueFrom(byteValue);
}
}
@SuppressWarnings("unchecked")
@Override
public synchronized void put(K key, V value) {
validateStoreOpen();
byte[] rawKey = serdes.rawKey(key);
byte[] rawValue = serdes.rawValue(value);
putInternal(rawKey, rawValue);
}
-Sameer.
Re: Synchronized methods in RockSB key store
Posted by Guozhang Wang <wa...@gmail.com>.
One rationale behind it is on the implementation of windowed store, which
may span over multiple RocksDB instances. When you have a range query over
a window store, we need to make sure that the underlying stores provide a
consistent "snapshot" at the time when the query is issued.
Such synchronization can be possibly lifted in the future, when we optimize
the window store implementation to not span over multiple RocksDB instances.
Guozhang
On Wed, Aug 16, 2017 at 4:19 AM, Sameer Kumar <sa...@gmail.com>
wrote:
> From the rocksdb writeup, it doesnt seem so. I am interested to know if
> there were any issues that we faced and added synchronization.
>
> -Sameer.
>
> On Wed, Aug 16, 2017 at 2:01 PM, Damian Guy <da...@gmail.com> wrote:
>
> > Sameer,
> > It might be that put, delete, putIfAbsent etc operations can be
> > non-synchronized. However for get and range operations that can be
> > performed by IQ, i.e, other threads, we need to guard against the store
> > being closed by the StreamThread, hence the synchronization.
> >
> > Thanks,
> > Damian
> >
> > On Wed, 16 Aug 2017 at 07:17 Sameer Kumar <sa...@gmail.com>
> wrote:
> >
> > > Can you share more info on this.
> > >
> > >
> > > As per the rocksb doc(
> > > https://github.com/facebook/rocksdb/wiki/basic-operations)
> > >
> > > Within a single process, the same rocksdb::DBobject may be safely
> shared
> > by
> > > multiple concurrent threads. I.e., different threads may write into or
> > > fetch iterators or call Get on the same database without any external
> > > synchronization (the leveldb implementation will automatically do the
> > > required synchronization).
> > >
> > > Did we faced any issues, I want to explore this and see if I can help
> on
> > > this. Is there a existing KIP on this.
> > >
> > > -Sameer.
> > >
> > >
> > > On Tue, Aug 15, 2017 at 10:42 AM, Sameer Kumar <sam.kum.work@gmail.com
> >
> > > wrote:
> > >
> > > > got it..Thanks Guozhang.
> > > >
> > > > On Tue, Aug 15, 2017 at 1:55 AM, Guozhang Wang <wa...@gmail.com>
> > > wrote:
> > > >
> > > >> Sameer,
> > > >>
> > > >> It is mainly to guard for concurrent access for interactive queries:
> > > >>
> > > >> https://kafka.apache.org/0110/documentation/streams/develope
> > > >> r-guide#streams_interactive_queries
> > > >>
> > > >> In Kafka Streams, we allow users to independently query the running
> > > state
> > > >> stores in real-time in their own caller thread while the
> application's
> > > >> thread is continuously updating these stores, and hence the
> > > >> synchronization.
> > > >>
> > > >>
> > > >> Guozhang
> > > >>
> > > >>
> > > >> On Sun, Aug 13, 2017 at 10:24 PM, Sameer Kumar <
> > sam.kum.work@gmail.com>
> > > >> wrote:
> > > >>
> > > >> > Hi,
> > > >> >
> > > >> > I was inspecting RocksDBStore and I saw get and put methods being
> > > >> > synchronized.
> > > >> >
> > > >> > Are there any issues in RocksDB due to which we have used it.
> Could
> > > >> someone
> > > >> > please enlighten more on this.
> > > >> >
> > > >> > @Override
> > > >> > public synchronized V get(K key) {
> > > >> > validateStoreOpen();
> > > >> > byte[] byteValue = getInternal(serdes.rawKey(key));
> > > >> > if (byteValue == null) {
> > > >> > return null;
> > > >> > } else {
> > > >> > return serdes.valueFrom(byteValue);
> > > >> > }
> > > >> > }
> > > >> >
> > > >> > @SuppressWarnings("unchecked")
> > > >> > @Override
> > > >> > public synchronized void put(K key, V value) {
> > > >> > validateStoreOpen();
> > > >> > byte[] rawKey = serdes.rawKey(key);
> > > >> > byte[] rawValue = serdes.rawValue(value);
> > > >> > putInternal(rawKey, rawValue);
> > > >> > }
> > > >> >
> > > >> > -Sameer.
> > > >> >
> > > >>
> > > >>
> > > >>
> > > >> --
> > > >> -- Guozhang
> > > >>
> > > >
> > > >
> > >
> >
>
--
-- Guozhang
Re: Synchronized methods in RockSB key store
Posted by Sameer Kumar <sa...@gmail.com>.
From the rocksdb writeup, it doesnt seem so. I am interested to know if
there were any issues that we faced and added synchronization.
-Sameer.
On Wed, Aug 16, 2017 at 2:01 PM, Damian Guy <da...@gmail.com> wrote:
> Sameer,
> It might be that put, delete, putIfAbsent etc operations can be
> non-synchronized. However for get and range operations that can be
> performed by IQ, i.e, other threads, we need to guard against the store
> being closed by the StreamThread, hence the synchronization.
>
> Thanks,
> Damian
>
> On Wed, 16 Aug 2017 at 07:17 Sameer Kumar <sa...@gmail.com> wrote:
>
> > Can you share more info on this.
> >
> >
> > As per the rocksb doc(
> > https://github.com/facebook/rocksdb/wiki/basic-operations)
> >
> > Within a single process, the same rocksdb::DBobject may be safely shared
> by
> > multiple concurrent threads. I.e., different threads may write into or
> > fetch iterators or call Get on the same database without any external
> > synchronization (the leveldb implementation will automatically do the
> > required synchronization).
> >
> > Did we faced any issues, I want to explore this and see if I can help on
> > this. Is there a existing KIP on this.
> >
> > -Sameer.
> >
> >
> > On Tue, Aug 15, 2017 at 10:42 AM, Sameer Kumar <sa...@gmail.com>
> > wrote:
> >
> > > got it..Thanks Guozhang.
> > >
> > > On Tue, Aug 15, 2017 at 1:55 AM, Guozhang Wang <wa...@gmail.com>
> > wrote:
> > >
> > >> Sameer,
> > >>
> > >> It is mainly to guard for concurrent access for interactive queries:
> > >>
> > >> https://kafka.apache.org/0110/documentation/streams/develope
> > >> r-guide#streams_interactive_queries
> > >>
> > >> In Kafka Streams, we allow users to independently query the running
> > state
> > >> stores in real-time in their own caller thread while the application's
> > >> thread is continuously updating these stores, and hence the
> > >> synchronization.
> > >>
> > >>
> > >> Guozhang
> > >>
> > >>
> > >> On Sun, Aug 13, 2017 at 10:24 PM, Sameer Kumar <
> sam.kum.work@gmail.com>
> > >> wrote:
> > >>
> > >> > Hi,
> > >> >
> > >> > I was inspecting RocksDBStore and I saw get and put methods being
> > >> > synchronized.
> > >> >
> > >> > Are there any issues in RocksDB due to which we have used it. Could
> > >> someone
> > >> > please enlighten more on this.
> > >> >
> > >> > @Override
> > >> > public synchronized V get(K key) {
> > >> > validateStoreOpen();
> > >> > byte[] byteValue = getInternal(serdes.rawKey(key));
> > >> > if (byteValue == null) {
> > >> > return null;
> > >> > } else {
> > >> > return serdes.valueFrom(byteValue);
> > >> > }
> > >> > }
> > >> >
> > >> > @SuppressWarnings("unchecked")
> > >> > @Override
> > >> > public synchronized void put(K key, V value) {
> > >> > validateStoreOpen();
> > >> > byte[] rawKey = serdes.rawKey(key);
> > >> > byte[] rawValue = serdes.rawValue(value);
> > >> > putInternal(rawKey, rawValue);
> > >> > }
> > >> >
> > >> > -Sameer.
> > >> >
> > >>
> > >>
> > >>
> > >> --
> > >> -- Guozhang
> > >>
> > >
> > >
> >
>
Re: Synchronized methods in RockSB key store
Posted by Damian Guy <da...@gmail.com>.
Sameer,
It might be that put, delete, putIfAbsent etc operations can be
non-synchronized. However for get and range operations that can be
performed by IQ, i.e, other threads, we need to guard against the store
being closed by the StreamThread, hence the synchronization.
Thanks,
Damian
On Wed, 16 Aug 2017 at 07:17 Sameer Kumar <sa...@gmail.com> wrote:
> Can you share more info on this.
>
>
> As per the rocksb doc(
> https://github.com/facebook/rocksdb/wiki/basic-operations)
>
> Within a single process, the same rocksdb::DBobject may be safely shared by
> multiple concurrent threads. I.e., different threads may write into or
> fetch iterators or call Get on the same database without any external
> synchronization (the leveldb implementation will automatically do the
> required synchronization).
>
> Did we faced any issues, I want to explore this and see if I can help on
> this. Is there a existing KIP on this.
>
> -Sameer.
>
>
> On Tue, Aug 15, 2017 at 10:42 AM, Sameer Kumar <sa...@gmail.com>
> wrote:
>
> > got it..Thanks Guozhang.
> >
> > On Tue, Aug 15, 2017 at 1:55 AM, Guozhang Wang <wa...@gmail.com>
> wrote:
> >
> >> Sameer,
> >>
> >> It is mainly to guard for concurrent access for interactive queries:
> >>
> >> https://kafka.apache.org/0110/documentation/streams/develope
> >> r-guide#streams_interactive_queries
> >>
> >> In Kafka Streams, we allow users to independently query the running
> state
> >> stores in real-time in their own caller thread while the application's
> >> thread is continuously updating these stores, and hence the
> >> synchronization.
> >>
> >>
> >> Guozhang
> >>
> >>
> >> On Sun, Aug 13, 2017 at 10:24 PM, Sameer Kumar <sa...@gmail.com>
> >> wrote:
> >>
> >> > Hi,
> >> >
> >> > I was inspecting RocksDBStore and I saw get and put methods being
> >> > synchronized.
> >> >
> >> > Are there any issues in RocksDB due to which we have used it. Could
> >> someone
> >> > please enlighten more on this.
> >> >
> >> > @Override
> >> > public synchronized V get(K key) {
> >> > validateStoreOpen();
> >> > byte[] byteValue = getInternal(serdes.rawKey(key));
> >> > if (byteValue == null) {
> >> > return null;
> >> > } else {
> >> > return serdes.valueFrom(byteValue);
> >> > }
> >> > }
> >> >
> >> > @SuppressWarnings("unchecked")
> >> > @Override
> >> > public synchronized void put(K key, V value) {
> >> > validateStoreOpen();
> >> > byte[] rawKey = serdes.rawKey(key);
> >> > byte[] rawValue = serdes.rawValue(value);
> >> > putInternal(rawKey, rawValue);
> >> > }
> >> >
> >> > -Sameer.
> >> >
> >>
> >>
> >>
> >> --
> >> -- Guozhang
> >>
> >
> >
>
Re: Synchronized methods in RockSB key store
Posted by Sameer Kumar <sa...@gmail.com>.
Can you share more info on this.
As per the rocksb doc(
https://github.com/facebook/rocksdb/wiki/basic-operations)
Within a single process, the same rocksdb::DBobject may be safely shared by
multiple concurrent threads. I.e., different threads may write into or
fetch iterators or call Get on the same database without any external
synchronization (the leveldb implementation will automatically do the
required synchronization).
Did we faced any issues, I want to explore this and see if I can help on
this. Is there a existing KIP on this.
-Sameer.
On Tue, Aug 15, 2017 at 10:42 AM, Sameer Kumar <sa...@gmail.com>
wrote:
> got it..Thanks Guozhang.
>
> On Tue, Aug 15, 2017 at 1:55 AM, Guozhang Wang <wa...@gmail.com> wrote:
>
>> Sameer,
>>
>> It is mainly to guard for concurrent access for interactive queries:
>>
>> https://kafka.apache.org/0110/documentation/streams/develope
>> r-guide#streams_interactive_queries
>>
>> In Kafka Streams, we allow users to independently query the running state
>> stores in real-time in their own caller thread while the application's
>> thread is continuously updating these stores, and hence the
>> synchronization.
>>
>>
>> Guozhang
>>
>>
>> On Sun, Aug 13, 2017 at 10:24 PM, Sameer Kumar <sa...@gmail.com>
>> wrote:
>>
>> > Hi,
>> >
>> > I was inspecting RocksDBStore and I saw get and put methods being
>> > synchronized.
>> >
>> > Are there any issues in RocksDB due to which we have used it. Could
>> someone
>> > please enlighten more on this.
>> >
>> > @Override
>> > public synchronized V get(K key) {
>> > validateStoreOpen();
>> > byte[] byteValue = getInternal(serdes.rawKey(key));
>> > if (byteValue == null) {
>> > return null;
>> > } else {
>> > return serdes.valueFrom(byteValue);
>> > }
>> > }
>> >
>> > @SuppressWarnings("unchecked")
>> > @Override
>> > public synchronized void put(K key, V value) {
>> > validateStoreOpen();
>> > byte[] rawKey = serdes.rawKey(key);
>> > byte[] rawValue = serdes.rawValue(value);
>> > putInternal(rawKey, rawValue);
>> > }
>> >
>> > -Sameer.
>> >
>>
>>
>>
>> --
>> -- Guozhang
>>
>
>
Re: Synchronized methods in RockSB key store
Posted by Sameer Kumar <sa...@gmail.com>.
got it..Thanks Guozhang.
On Tue, Aug 15, 2017 at 1:55 AM, Guozhang Wang <wa...@gmail.com> wrote:
> Sameer,
>
> It is mainly to guard for concurrent access for interactive queries:
>
> https://kafka.apache.org/0110/documentation/streams/
> developer-guide#streams_interactive_queries
>
> In Kafka Streams, we allow users to independently query the running state
> stores in real-time in their own caller thread while the application's
> thread is continuously updating these stores, and hence the
> synchronization.
>
>
> Guozhang
>
>
> On Sun, Aug 13, 2017 at 10:24 PM, Sameer Kumar <sa...@gmail.com>
> wrote:
>
> > Hi,
> >
> > I was inspecting RocksDBStore and I saw get and put methods being
> > synchronized.
> >
> > Are there any issues in RocksDB due to which we have used it. Could
> someone
> > please enlighten more on this.
> >
> > @Override
> > public synchronized V get(K key) {
> > validateStoreOpen();
> > byte[] byteValue = getInternal(serdes.rawKey(key));
> > if (byteValue == null) {
> > return null;
> > } else {
> > return serdes.valueFrom(byteValue);
> > }
> > }
> >
> > @SuppressWarnings("unchecked")
> > @Override
> > public synchronized void put(K key, V value) {
> > validateStoreOpen();
> > byte[] rawKey = serdes.rawKey(key);
> > byte[] rawValue = serdes.rawValue(value);
> > putInternal(rawKey, rawValue);
> > }
> >
> > -Sameer.
> >
>
>
>
> --
> -- Guozhang
>
Re: Synchronized methods in RockSB key store
Posted by Guozhang Wang <wa...@gmail.com>.
Sameer,
It is mainly to guard for concurrent access for interactive queries:
https://kafka.apache.org/0110/documentation/streams/developer-guide#streams_interactive_queries
In Kafka Streams, we allow users to independently query the running state
stores in real-time in their own caller thread while the application's
thread is continuously updating these stores, and hence the synchronization.
Guozhang
On Sun, Aug 13, 2017 at 10:24 PM, Sameer Kumar <sa...@gmail.com>
wrote:
> Hi,
>
> I was inspecting RocksDBStore and I saw get and put methods being
> synchronized.
>
> Are there any issues in RocksDB due to which we have used it. Could someone
> please enlighten more on this.
>
> @Override
> public synchronized V get(K key) {
> validateStoreOpen();
> byte[] byteValue = getInternal(serdes.rawKey(key));
> if (byteValue == null) {
> return null;
> } else {
> return serdes.valueFrom(byteValue);
> }
> }
>
> @SuppressWarnings("unchecked")
> @Override
> public synchronized void put(K key, V value) {
> validateStoreOpen();
> byte[] rawKey = serdes.rawKey(key);
> byte[] rawValue = serdes.rawValue(value);
> putInternal(rawKey, rawValue);
> }
>
> -Sameer.
>
--
-- Guozhang