apoorvmittal10 commented on code in PR #17965:
URL: https://github.com/apache/kafka/pull/17965#discussion_r1868535050
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
Review Comment:
Seems the whole purpose exposing `rollbackState` is to verify if there is an
ongoing transaction, correct? Then why to have state exposed rather have a
method here `hasOngoingStateTransition()` which should return a boolean
specifying if the state has any transaction in progress.
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't
transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask
acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask
acquisitionLockTimeoutTask) throws IllegalStateException {
Review Comment:
Is it really a IllesgalStateException or IllegalArgumentException for the
method?
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2332,12 +2330,24 @@ String memberId() {
return memberId;
}
+ private RecordState rollbackState() {
+ if (rollbackState == null) {
+ // This case could occur when the batch/offset hasn't
transitioned even once or the state transitions have
+ // been committed.
+ return null;
+ }
+ return rollbackState.state;
+ }
+
// Visible for testing.
TimerTask acquisitionLockTimeoutTask() {
return acquisitionLockTimeoutTask;
}
- void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask
acquisitionLockTimeoutTask) {
+ void updateAcquisitionLockTimeoutTask(AcquisitionLockTimerTask
acquisitionLockTimeoutTask) throws IllegalStateException {
+ if (this.acquisitionLockTimeoutTask != null) {
+ throw new IllegalStateException("The acquisition lock timeout
task is being overridden");
Review Comment:
`"The acquisition lock timeout task is being overridden"` I didn't
understanf the error message on `null` `acquisitionLockTimeoutTask`
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2189,12 +2188,16 @@ long lastOffset() {
return lastOffset;
}
- // Visible for testing.
- RecordState batchState() {
+ private InFlightState inFlightState() {
Review Comment:
nit: move private methods in the end i.e. after public -> protected ->
default -> private
##########
core/src/main/java/kafka/server/share/SharePartition.java:
##########
@@ -2189,12 +2188,16 @@ long lastOffset() {
return lastOffset;
}
- // Visible for testing.
- RecordState batchState() {
+ private InFlightState inFlightState() {
if (batchState == null) {
throw new IllegalStateException("The batch state is not
available as the offset state is maintained");
}
- return batchState.state;
+ return batchState;
+ }
+
+ // Visible for testing.
+ RecordState batchState() {
Review Comment:
I didn't get the intention of this change, how it's helpful?
--
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]
