[jira] [Commented] (KAFKA-6412) Improve synchronization in CachingKeyValueStore methods

[Guozhang Wang (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-6412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16320898#comment-16320898
 ] 

Guozhang Wang commented on KAFKA-6412:
--

[~tedyu] way to start 2018 indeed :)

> Improve synchronization in CachingKeyValueStore methods
> ---
>
> Key: KAFKA-6412
> URL: https://issues.apache.org/jira/browse/KAFKA-6412
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Ted Yu
>Assignee: Ted Yu
> Fix For: 1.1.0
>
> Attachments: 6412-jmh.v1.txt, k-6412.v1.txt
>
>
> Currently CachingKeyValueStore methods are synchronized at method level.
> It seems we can use read lock for getter and write lock for put / delete 
> methods.
> For getInternal(), if the underlying thread is streamThread, the 
> getInternal() may trigger eviction. This can be handled by obtaining write 
> lock at the beginning of the method for streamThread.



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

[jira] [Commented] (KAFKA-6412) Improve synchronization in CachingKeyValueStore methods

[Ted Yu (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-6412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16320047#comment-16320047
 ] 

Ted Yu commented on KAFKA-6412:
---

I came up with the initial idea for this JIRA when sitting in hotel lobby at 
Grand Canyon.

It was nice way to start 2018.

> Improve synchronization in CachingKeyValueStore methods
> ---
>
> Key: KAFKA-6412
> URL: https://issues.apache.org/jira/browse/KAFKA-6412
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Ted Yu
> Fix For: 1.1.0
>
> Attachments: 6412-jmh.v1.txt, k-6412.v1.txt
>
>
> Currently CachingKeyValueStore methods are synchronized at method level.
> It seems we can use read lock for getter and write lock for put / delete 
> methods.
> For getInternal(), if the underlying thread is streamThread, the 
> getInternal() may trigger eviction. This can be handled by obtaining write 
> lock at the beginning of the method for streamThread.



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

[jira] [Commented] (KAFKA-6412) Improve synchronization in CachingKeyValueStore methods

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-6412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16320013#comment-16320013
 ] 

ASF GitHub Bot commented on KAFKA-6412:
---

dguy closed pull request #4372: KAFKA-6412 Improve synchronization in 
CachingKeyValueStore methods
URL: https://github.com/apache/kafka/pull/4372
 
 
   

This is a PR merged from a forked repository.
As GitHub hides the original diff on merge, it is displayed below for
the sake of provenance:

As this is a foreign pull request (from a fork), the diff is supplied
below (as it won't show otherwise due to GitHub magic):

diff --git 
a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
 
b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
index f0669a4f6ee..9fff8ccca04 100644
--- 
a/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
+++ 
b/streams/src/main/java/org/apache/kafka/streams/state/internals/CachingKeyValueStore.java
@@ -31,6 +31,9 @@
 
 import java.util.List;
 import java.util.Objects;
+import java.util.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReadWriteLock;
+import java.util.concurrent.locks.ReentrantReadWriteLock;
 
 class CachingKeyValueStore extends WrappedStateStore.AbstractStateStore 
implements KeyValueStore, CachedStateStore {
 
@@ -44,6 +47,7 @@
 private InternalProcessorContext context;
 private StateSerdes serdes;
 private Thread streamThread;
+private ReadWriteLock lock = new ReentrantReadWriteLock();
 
 CachingKeyValueStore(final KeyValueStore underlying,
  final Serde keySerde,
@@ -108,9 +112,14 @@ public void setFlushListener(final CacheFlushListener flushListener,
 }
 
 @Override
-public synchronized void flush() {
-cache.flush(cacheName);
-underlying.flush();
+public void flush() {
+lock.writeLock().lock();
+try {
+cache.flush(cacheName);
+underlying.flush();
+} finally {
+lock.writeLock().unlock();
+}
 }
 
 @Override
@@ -131,10 +140,21 @@ public boolean isOpen() {
 }
 
 @Override
-public synchronized byte[] get(final Bytes key) {
+public byte[] get(final Bytes key) {
 validateStoreOpen();
-Objects.requireNonNull(key);
-return getInternal(key);
+Lock theLock;
+if (Thread.currentThread().equals(streamThread)) {
+theLock = lock.writeLock();
+} else {
+theLock = lock.readLock();
+}
+theLock.lock();
+try {
+Objects.requireNonNull(key);
+return getInternal(key);
+} finally {
+theLock.unlock();
+}
 }
 
 private byte[] getInternal(final Bytes key) {
@@ -176,50 +196,75 @@ public boolean isOpen() {
 }
 
 @Override
-public synchronized long approximateNumEntries() {
+public long approximateNumEntries() {
 validateStoreOpen();
-return underlying.approximateNumEntries();
+lock.readLock().lock();
+try {
+return underlying.approximateNumEntries();
+} finally {
+lock.readLock().unlock();
+}
 }
 
 @Override
-public synchronized void put(final Bytes key, final byte[] value) {
+public void put(final Bytes key, final byte[] value) {
 Objects.requireNonNull(key, "key cannot be null");
 validateStoreOpen();
-putInternal(key, value);
+lock.writeLock().lock();
+try {
+putInternal(key, value);
+} finally {
+lock.writeLock().unlock();
+}
 }
 
-private synchronized void putInternal(final Bytes rawKey, final byte[] 
value) {
+private void putInternal(final Bytes rawKey, final byte[] value) {
 Objects.requireNonNull(rawKey, "key cannot be null");
 cache.put(cacheName, rawKey, new LRUCacheEntry(value, true, 
context.offset(),
-  context.timestamp(), context.partition(), context.topic()));
+  context.timestamp(), context.partition(), context.topic()));
 }
 
 @Override
-public synchronized byte[] putIfAbsent(final Bytes key, final byte[] 
value) {
+public byte[] putIfAbsent(final Bytes key, final byte[] value) {
 Objects.requireNonNull(key, "key cannot be null");
 validateStoreOpen();
-final byte[] v = getInternal(key);
-if (v == null) {
-putInternal(key, value);
+lock.writeLock().lock();
+try {
+final byte[] v = getInternal(key);
+if (v == null) {
+putInternal(key, value);
+}
+return v;
+} finally {
+lock.writeLock().unlock();
 }
-return v;
 }
 
 @Override
- 

[jira] [Commented] (KAFKA-6412) Improve synchronization in CachingKeyValueStore methods

[ASF GitHub Bot (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-6412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16307468#comment-16307468
 ] 

ASF GitHub Bot commented on KAFKA-6412:
---

tedyu opened a new pull request #4372: KAFKA-6412 Improve synchronization in 
CachingKeyValueStore methods
URL: https://github.com/apache/kafka/pull/4372
 
 
   Currently CachingKeyValueStore methods are synchronized at method level.
   
   It seems we can use read lock for getter and write lock for put / delete 
methods.
   
   For getInternal(), if the underlying thread is streamThread, the 
getInternal() may trigger eviction. This can be handled by obtaining write lock 
at the beginning of the method for streamThread.
   
   ### Committer Checklist (excluded from commit message)
   - [ ] Verify design and implementation 
   - [ ] Verify test coverage and CI build status
   - [ ] Verify documentation (including upgrade notes)
   


This is an automated message from the Apache Git Service.
To respond to the message, please log on GitHub and use the
URL above to go to the specific comment.
 
For queries about this service, please contact Infrastructure at:
us...@infra.apache.org


> Improve synchronization in CachingKeyValueStore methods
> ---
>
> Key: KAFKA-6412
> URL: https://issues.apache.org/jira/browse/KAFKA-6412
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Ted Yu
> Attachments: k-6412.v1.txt
>
>
> Currently CachingKeyValueStore methods are synchronized at method level.
> It seems we can use read lock for getter and write lock for put / delete 
> methods.
> For getInternal(), if the underlying thread is streamThread, the 
> getInternal() may trigger eviction. This can be handled by obtaining write 
> lock at the beginning of the method for streamThread.



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

[jira] [Commented] (KAFKA-6412) Improve synchronization in CachingKeyValueStore methods

[Matthias J. Sax (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-6412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16306934#comment-16306934
 ] 

Matthias J. Sax commented on KAFKA-6412:


[~tedyu] Please open a PR on Github and we can take it from there. Thx.

> Improve synchronization in CachingKeyValueStore methods
> ---
>
> Key: KAFKA-6412
> URL: https://issues.apache.org/jira/browse/KAFKA-6412
> Project: Kafka
>  Issue Type: Improvement
>  Components: streams
>Reporter: Ted Yu
> Attachments: k-6412.v1.txt
>
>
> Currently CachingKeyValueStore methods are synchronized at method level.
> It seems we can use read lock for getter and write lock for put / delete 
> methods.
> For getInternal(), if the underlying thread is streamThread, the 
> getInternal() may trigger eviction. This can be handled by obtaining write 
> lock at the beginning of the method for streamThread.



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

[jira] [Commented] (KAFKA-6412) Improve synchronization in CachingKeyValueStore methods

[Ted Yu (JIRA)]

[ 
https://issues.apache.org/jira/browse/KAFKA-6412?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=16306689#comment-16306689
 ] 

Ted Yu commented on KAFKA-6412:
---

[~damianguy] [~guozhang] [~mjsax] :
See if I missed something.

> Improve synchronization in CachingKeyValueStore methods
> ---
>
> Key: KAFKA-6412
> URL: https://issues.apache.org/jira/browse/KAFKA-6412
> Project: Kafka
>  Issue Type: Improvement
>Reporter: Ted Yu
>
> Currently CachingKeyValueStore methods are synchronized at method level.
> It seems we can use read lock for getter and write lock for put / delete 
> methods.
> For getInternal(), if the underlying thread is streamThread, the 
> getInternal() may trigger eviction. This can be handled by obtaining write 
> lock at the beginning of the method for streamThread.



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