bbejeck commented on code in PR #22683:
URL: https://github.com/apache/kafka/pull/22683#discussion_r3482951097
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java:
##########
@@ -1479,7 +1523,21 @@ public Options getOptions() {
@Override
public Position getPosition() {
- return position;
+ synchronized (position) {
+ return
position.copy().merge(dbAccessor.uncommittedPositionDeltas());
+ }
+ }
+
+ // Visible for testing. Returns a snapshot of the committed Position only
(no pending writes).
+ Position committedPositionForTest() {
Review Comment:
Is the sole purpose of this method and the one below for testing? Is there
an existing method that is currently `private` we could open up to `protected `
instead?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -272,8 +279,42 @@ void flushToBase() {
@Override
void discardPendingBatch() {
+ // Called from rollback() under the snapshotLock write lock: discard
the staged position
Review Comment:
Not sure we need these two comments
##########
streams/src/test/java/org/apache/kafka/streams/state/internals/RocksDBStoreTest.java:
##########
@@ -1519,6 +1519,55 @@ public void
readOnlyOnNonTransactionalStoreShouldBehaveIdenticallyAcrossLevels()
rocksDBStore.readOnly(IsolationLevel.READ_COMMITTED).get(key)));
}
+ @Test
Review Comment:
The test doesn't cover the rollback path (`discardPendingBatch()` clearing
pendingPosition) - can we add a couple for that path?
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -272,8 +279,42 @@ void flushToBase() {
@Override
void discardPendingBatch() {
+ // Called from rollback() under the snapshotLock write lock: discard
the staged position
+ // alongside the staged writes and range tombstones.
writeBatch.clear();
rangeTombstones = Collections.emptyNavigableMap();
+ pendingPosition = Position.emptyPosition();
+ }
+
+ /** Stage the current record's position into the uncommitted transaction
(owner write path). */
Review Comment:
same here
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBTransactionBuffer.java:
##########
@@ -272,8 +279,42 @@ void flushToBase() {
@Override
void discardPendingBatch() {
+ // Called from rollback() under the snapshotLock write lock: discard
the staged position
+ // alongside the staged writes and range tombstones.
writeBatch.clear();
rangeTombstones = Collections.emptyNavigableMap();
+ pendingPosition = Position.emptyPosition();
+ }
+
+ /** Stage the current record's position into the uncommitted transaction
(owner write path). */
+ void updatePosition(final StateStoreContext stateStoreContext) {
+ snapshotLock.writeLock().lock();
+ try {
+ StoreQueryUtils.updatePosition(pendingPosition, stateStoreContext);
+ } finally {
+ snapshotLock.writeLock().unlock();
+ }
+ }
+
+ /** A point-in-time copy of the uncommitted position deltas, for
isolation-aware (IQ) reads. */
Review Comment:
ditto - pretty much the same for all these comments unless explaining why
something is done
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/RocksDBStore.java:
##########
@@ -531,12 +531,20 @@ public <R> QueryResult<R> query(
final PositionBound positionBound,
final QueryConfig config) {
+ final Position queryPosition;
+ synchronized (position) {
+ if (config.getIsolationLevel() == IsolationLevel.READ_COMMITTED) {
+ queryPosition = position.copy();
Review Comment:
II think this might be an issue. `position.copy()` returns a fresh instance
that's local to this call, and it's what gets passed to
`StoreQueryUtils.handleBasicQueries`, where it's used as the monitor on line
131. Since no other thread can reference that instance, the lock there
isn't really guarding anything — in particular it no longer excludes a
concurrent put()/commit() on the processor thread the way it did when the
store's own position was passed in.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
