)]}' { "commit": "c8374348e5c23c72ac276163be6ac123b1b3bae2", "tree": "43f23478438caa7fcc7b44ab8e17c7086cbb9025", "parents": [ "959f9ca4c02817cce85f0375d431ed72f22699e2" ], "author": { "name": "Colin P. McCabe", "email": "cmccabe@apache.org", "time": "Tue Apr 25 16:09:35 2023 -0700" }, "committer": { "name": "Colin P. McCabe", "email": "cmccabe@apache.org", "time": "Thu Jul 13 13:46:56 2023 -0700" }, "message": "MINOR: provide the exact offset to QuorumController.replay\n\nProvide the exact record offset to QuorumController.replay() in all cases. There are several situations\nwhere this is useful, such as logging, implementing metadata transactions, or handling broker\nregistration records.\n\nIn the case where the QC is inactive, and simply replaying records, it is easy to compute the exact\nrecord offset from the batch base offset and the record index.\n\nThe active QC case is more difficult. Technically, when we submit records to the Raft layer, it can\nchoose a batch end offset later than the one we expect, if someone else is also adding records.\nWhile the QC is the only entity submitting data records, control records may be added at any time.\nIn the current implementation, these are really only used for leadership elections. However, this\ncould change with the addition of quorum reconfiguration or similar features.\n\nTherefore, this PR allows the QC to tell the Raft layer that a record append should fail if it\nwould have resulted in a batch end offset other than what was expected. This in turn will trigger a\ncontroller failover. In the future, if automatically added control records become more common, we\nmay wish to have a more sophisticated system than this simple optimistic concurrency mechanism. But\nfor now, this will allow us to rely on the offset as correct.\n\nIn order that the active QC can learn what offset to start writing at, the PR also adds a new\nendOffset parameter to handleLeaderChange. Since the Raft layer only invokes handleLeaderChange on\nthe active once it has replayed the log, this information should always be up-to-date in that\ncontext.\n\nAt the Raft level, this PR adds a new exception, UnexpectedEndOffsetException. This gets thrown\nwhen we request an end offset that doesn\u0027t match the one the Raft layer would have given us.\nAlthough this exception should cause a failover, it should not be considered a fault. This\ncomplicated the exception handling a bit and motivated splitting more of it out into the new\nEventHandlerExceptionInfo class. This will also let us unit test things like slf4j log messages a\nbit better.\n", "tree_diff": [ { "type": "modify", "old_id": "1026c528473370ef371b942da2a78efb8c6e64ea", "old_mode": 33188, "old_path": "core/src/main/scala/kafka/tools/TestRaftServer.scala", "new_id": "6d2653979ada063b0465741731cc8e2f9c47ceea", "new_mode": 33188, "new_path": "core/src/main/scala/kafka/tools/TestRaftServer.scala" }, { "type": "modify", "old_id": "c252afdd6ec159120325700e07d061e79c461a30", "old_mode": 33188, "old_path": "metadata/src/main/java/org/apache/kafka/controller/QuorumController.java", "new_id": "5bf528451f041a13a8f5973d56aca5e2fa0fa037", "new_mode": 33188, "new_path": "metadata/src/main/java/org/apache/kafka/controller/QuorumController.java" }, { "type": "modify", "old_id": "6f99ea45f6b5dfb9b1ddd1733dd617e0197db218", "old_mode": 33188, "old_path": "metadata/src/main/java/org/apache/kafka/controller/errors/ControllerExceptions.java", "new_id": "b7e74446a4b6be901b08e8e17f7bdc2e7279181e", "new_mode": 33188, "new_path": "metadata/src/main/java/org/apache/kafka/controller/errors/ControllerExceptions.java" }, { "type": "add", "old_id": "0000000000000000000000000000000000000000", "old_mode": 0, "old_path": "/dev/null", "new_id": "4e74b71a677b0d75e3fabe94142a6a80410b25f6", "new_mode": 33188, "new_path": "metadata/src/main/java/org/apache/kafka/controller/errors/EventHandlerExceptionInfo.java" }, { "type": "modify", "old_id": "768fcb2574b2fa8ff05d3e48498f9d1caefc1b39", "old_mode": 33188, "old_path": "metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java", "new_id": "c923742af9fa7a427558c640b9a5134071f79789", "new_mode": 33188, "new_path": "metadata/src/main/java/org/apache/kafka/image/loader/MetadataLoader.java" }, { "type": "modify", "old_id": "6f82d915d8d5dd40a904419d260f25c916710b6c", "old_mode": 33188, "old_path": "metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileWriter.java", "new_id": "0ccf39a85ac8fdcc779f6c046b22ad07b727cd2a", "new_mode": 33188, "new_path": "metadata/src/main/java/org/apache/kafka/metadata/util/BatchFileWriter.java" }, { "type": "modify", "old_id": "19875741ec2dea0622df26ebe67862629053361a", "old_mode": 33188, "old_path": "metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotFileReader.java", "new_id": "5db1e575e796753fcc466d477c664e86cb5c9cde", "new_mode": 33188, "new_path": "metadata/src/main/java/org/apache/kafka/metadata/util/SnapshotFileReader.java" }, { "type": "modify", "old_id": "a440ce53122a717b960c93b4c4aa81fdfa6d03b6", "old_mode": 33188, "old_path": "metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java", "new_id": "41bffdba92fc6cfe9e822f252f095c424eaafeac", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/controller/ReplicationControlManagerTest.java" }, { "type": "modify", "old_id": "81c234491b5584e1837b3617ff2ed587f5bd876d", "old_mode": 33188, "old_path": "metadata/src/test/java/org/apache/kafka/controller/errors/ControllerExceptionsTest.java", "new_id": "2d1905b1a5c1cde2d865dc355fa1e687f109e93c", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/controller/errors/ControllerExceptionsTest.java" }, { "type": "add", "old_id": "0000000000000000000000000000000000000000", "old_mode": 0, "old_path": "/dev/null", "new_id": "ccffde2831e1fced25ed8701e251bd6554c64739", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/controller/errors/EventHandlerExceptionInfoTest.java" }, { "type": "modify", "old_id": "ca72aa058ef9da8143c617dddf4fa2b46e610c17", "old_mode": 33188, "old_path": "metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java", "new_id": "038de599009f2e1f2dcd61f6fbb996a47ee6c1a4", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/image/publisher/SnapshotEmitterTest.java" }, { "type": "modify", "old_id": "46f99db5d1d1eb2981e8d8404272c450051ec0ab", "old_mode": 33188, "old_path": "metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java", "new_id": "1f1b5d72e4754127e4b620c6d6ead389aa9fe0b6", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/metalog/LocalLogManager.java" }, { "type": "modify", "old_id": "d72b7557b48e0dd5f8f1abbc47446894480a8f88", "old_mode": 33188, "old_path": "metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java", "new_id": "c3a7283ab9edceb277004ce39d98224301aadccc", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/metalog/LocalLogManagerTestEnv.java" }, { "type": "modify", "old_id": "3d7267d94a5ee81165ecd553ec03d122d4d2da38", "old_mode": 33188, "old_path": "metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java", "new_id": "026f0ac530191ef6f665b448bb4652c0d2389cef", "new_mode": 33188, "new_path": "metadata/src/test/java/org/apache/kafka/metalog/MockMetaLogManagerListener.java" }, { "type": "modify", "old_id": "40c6a695908045ddf5bbb0dd0f97960b011d1da9", "old_mode": 33188, "old_path": "raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java", "new_id": "764c0999558c90b09b513743f15a8c95151c8f54", "new_mode": 33188, "new_path": "raft/src/main/java/org/apache/kafka/raft/KafkaRaftClient.java" }, { "type": "modify", "old_id": "6422fb5f3a095a09f54cc31ff3dca20eefcaa218", "old_mode": 33188, "old_path": "raft/src/main/java/org/apache/kafka/raft/RaftClient.java", "new_id": "2b899923f91cbfbd9e26ef43af804d925117322e", "new_mode": 33188, "new_path": "raft/src/main/java/org/apache/kafka/raft/RaftClient.java" }, { "type": "modify", "old_id": "cceb65930edbfa47585523c27e2b85e0b43a60bd", "old_mode": 33188, "old_path": "raft/src/main/java/org/apache/kafka/raft/ReplicatedCounter.java", "new_id": "36fa99e12c910514082e705342f8e2b93f67a4cd", "new_mode": 33188, "new_path": "raft/src/main/java/org/apache/kafka/raft/ReplicatedCounter.java" }, { "type": "add", "old_id": "0000000000000000000000000000000000000000", "old_mode": 0, "old_path": "/dev/null", "new_id": "f470134c9a3767684dbf6ca328b53e17142bb9bb", "new_mode": 33188, "new_path": "raft/src/main/java/org/apache/kafka/raft/errors/UnexpectedEndOffsetException.java" }, { "type": "modify", "old_id": "b84a7d57b8a7234b8c63d004fd2c7b24aa1146a5", "old_mode": 33188, "old_path": "raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java", "new_id": "87717d206d41ca6b0ed0e75e23da685702ba60b9", "new_mode": 33188, "new_path": "raft/src/main/java/org/apache/kafka/raft/internals/BatchAccumulator.java" }, { "type": "modify", "old_id": "eeacf608a9ff095ba676941a361a079924cb8630", "old_mode": 33188, "old_path": "raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java", "new_id": "a3f726be63e466abc023671798785dbbee05d4d1", "new_mode": 33188, "new_path": "raft/src/main/java/org/apache/kafka/snapshot/RecordsSnapshotWriter.java" }, { "type": "modify", "old_id": "7cbeb11f4f780db26a4d2c2ecb137fb1e1e781a3", "old_mode": 33188, "old_path": "raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java", "new_id": "44704e901f6859c2dc79a50c919e9fd9266a334a", "new_mode": 33188, "new_path": "raft/src/test/java/org/apache/kafka/raft/KafkaRaftClientTest.java" }, { "type": "modify", "old_id": "06a300f666c0f282142f0ec513ad49ed1aeb48ba", "old_mode": 33188, "old_path": "raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java", "new_id": "6f8e3e422e3045dce73b2e70e86648c1b3887de7", "new_mode": 33188, "new_path": "raft/src/test/java/org/apache/kafka/raft/RaftClientTestContext.java" }, { "type": "modify", "old_id": "11499771790f9fc2d65fca3684baa6a7841c39ae", "old_mode": 33188, "old_path": "raft/src/test/java/org/apache/kafka/raft/internals/BatchAccumulatorTest.java", "new_id": "ed7f6c94122f80808a9fd0f37e014d221184f44f", "new_mode": 33188, "new_path": "raft/src/test/java/org/apache/kafka/raft/internals/BatchAccumulatorTest.java" } ] }