THIS IS A TEST INSTANCE. ALL YOUR CHANGES WILL BE LOST!!!!
...
When one of the described checks in acquire or release fail, we throw a ConcurrentModificationException similar to current behavior of acquire and release.
Thread safety
Methods acquire and release need to make sure that memory writes from all thread involved are visible for each other. The pull request accomplishes this by using a synchronized block on a shared variable.
This is sufficient as can be read in the JSR-133 FAQ:
But there is more to synchronization than mutual exclusion. Synchronization ensures that memory writes by a thread before or during a synchronized block are made visible in a predictable manner to other threads which synchronize on the same monitor. After we exit a synchronized block, we release the monitor, which has the effect of flushing the cache to main memory, so that writes made by this thread can be visible to other threads. Before we can enter a synchronized block, we acquire the monitor, which has the effect of invalidating the local processor cache so that variables will be reloaded from main memory. We will then be able to see all of the writes made visible by the previous release.
For completeness, here follows the new proposed implementations of acquire and release.
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
// Holds the key that this thread needs to access the consumer, it is used to prevent multi-threaded access.
private final ThreadLocal<ThreadAccessKey> threadAccessKeyHolder = new ThreadLocal<>();
// The stack of allowed thread access keys. The top of the stack contains the access key of the thread that is
// currently allowed to use the consumer. When the stack is empty, any thread is allowed. Access is synchronized on
// the instance.
private final Deque<ThreadAccessKey> threadAccessStack = new ArrayDeque<>(4); |
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
private void acquire() {
final ThreadAccessKey threadAccessKey = threadAccessKeyHolder.get();
final ThreadAccessKey nextKey = new ThreadAccessKey();
synchronized (threadAccessStack) {
// Access is granted when threadAccess is empty (consumer is currently not used), or
// when the top value is the same as current key (consumer is used from callback)
if (threadAccessStack.isEmpty() || threadAccessStack.getFirst() == threadAccessKey) {
threadAccessKeyHolder.set(nextKey);
threadAccessStack.addFirst(nextKey);
} else {
final Thread thread = Thread.currentThread();
throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access. " +
"currentThread(name: " + thread.getName() + ", id: " + thread.getId() + ")" +
" could not provide access key (" + threadAccessStack.getFirst() + ")"
);
}
}
} |
| Code Block | ||||||||
|---|---|---|---|---|---|---|---|---|
| ||||||||
private void release() {
final ThreadAccessKey threadAccessKey = threadAccessKeyHolder.get();
synchronized (threadAccessStack) {
if (threadAccessStack.isEmpty()) {
throw new AssertionError("KafkaConsumer invariant violated: `release` invoked without `acquire`");
} else if (threadAccessStack.getFirst() == threadAccessKey) {
threadAccessStack.removeFirst();
if (threadAccessStack.isEmpty()) {
threadAccessKeyHolder.set(null);
} else {
threadAccessKeyHolder.set(threadAccessStack.getFirst());
}
} else {
final Thread thread = Thread.currentThread();
throw new ConcurrentModificationException("KafkaConsumer is not safe for multi-threaded access. " +
"currentThread(name: " + thread.getName() + ", id: " + thread.getId() + ")" +
" returned from callback but not provide access key (" + threadAccessStack.getFirst() + ")"
);
}
}
} |
Performance impact of the synchronized block is minimal because there will be no contention. Contention can only be caused by a badly written client and always results in a ConcurrentModificationException.
Compatibility, Deprecation, and Migration Plan
...