bbejeck commented on code in PR #22626:
URL: https://github.com/apache/kafka/pull/22626#discussion_r3454566968
##########
streams/src/main/java/org/apache/kafka/streams/state/internals/InMemoryKeyValueStore.java:
##########
@@ -231,13 +267,34 @@ public long approximateNumEntries() {
return map.size();
}
+ @Override
+ public long approximateNumUncommittedBytes() {
+ if (transactionBuffer != null) {
+ return transactionBuffer.approximateNumUncommittedBytes();
+ }
+ return 0;
+ }
+
@Override
public void commit(final Map<TopicPartition, Long> changelogOffsets) {
- // do-nothing since it is in-memory
+ commitStagedWrites();
+ }
+
+ void commitStagedWrites() {
+ if (transactionBuffer != null) {
+ transactionBuffer.commit();
+ }
+ }
+
+ void rollbackStagedWrites() {
+ if (transactionBuffer != null) {
+ transactionBuffer.rollback();
+ }
}
@Override
public void close() {
+ rollbackStagedWrites();
Review Comment:
Just a thought for discussion later - on `close` is there any chance we
would be stranding any writes - but I don't think so. I believe if the KS
producer had records to send and `KafkaStreams.close()` is executed and a
commit would eventually follow clearing the staged writes - speaking of the
happy-path only
--
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]
