You are viewing a plain text version of this content. The canonical link for it is here.
Posted to jira@kafka.apache.org by "kowshik (via GitHub)" <gi...@apache.org> on 2023/02/17 04:41:47 UTC
[GitHub] [kafka] kowshik opened a new pull request, #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
kowshik opened a new pull request, #13268:
URL: https://github.com/apache/kafka/pull/13268
Some of the `LeaderEndpoint` interface methods used a tuple of `(Int, Long)` as return value to represent epoch & offset. In this PR, I have replaced those occcurences with the existing `OffsetAndEpoch` type. This should hopefully improve readability, and eliminate the chances of developer confusing offset with epoch at the call site or return site.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "kowshik (via GitHub)" <gi...@apache.org>.
kowshik commented on PR #13268:
URL: https://github.com/apache/kafka/pull/13268#issuecomment-1440982486
@junrao Thanks for the review. I have rebased the PR and brought in the latest commits from AK trunk.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "kowshik (via GitHub)" <gi...@apache.org>.
kowshik commented on PR #13268:
URL: https://github.com/apache/kafka/pull/13268#issuecomment-1437650383
@junrao Thanks for the review! I've addressed the comments, the PR is ready for another pass.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao commented on a diff in pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao commented on code in PR #13268:
URL: https://github.com/apache/kafka/pull/13268#discussion_r1110272453
##########
core/src/main/scala/kafka/server/OffsetAndEpoch.scala:
##########
@@ -0,0 +1,24 @@
+/**
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package kafka.server
+
+case class OffsetAndEpoch(offset: Long, leaderEpoch: Int) {
Review Comment:
Could we add this in the server-common module in java?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a diff in pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "kowshik (via GitHub)" <gi...@apache.org>.
kowshik commented on code in PR #13268:
URL: https://github.com/apache/kafka/pull/13268#discussion_r1113595191
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -704,18 +708,18 @@ abstract class AbstractFetcherThread(name: String,
* Putting the two cases together, the follower should fetch from the higher one of its replica log end offset
* and the current leader's (local-log-start-offset or) log start offset.
*/
- val (epoch, leaderStartOffset) = if (fetchFromLocalLogStartOffset)
+ val offsetAndEpoch = if (fetchFromLocalLogStartOffset)
leader.fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch) else
leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)
-
+ val leaderStartOffset = offsetAndEpoch.offset
warn(s"Reset fetch offset for partition $topicPartition from $replicaEndOffset to current " +
s"leader's start offset $leaderStartOffset")
val offsetToFetch =
if (leaderStartOffset > replicaEndOffset) {
// Only truncate log when current leader's log start offset (local log start offset if >= 3.4 version incaseof
// OffsetMovedToTieredStorage error) is greater than follower's log end offset.
// truncateAndBuild returns offset value from which it needs to start fetching.
- truncateAndBuild(epoch, leaderStartOffset)
+ truncateAndBuild(offsetAndEpoch.leaderEpoch, leaderStartOffset)
Review Comment:
Done in 3da1e63. Good point!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Hangleton commented on a diff in pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on code in PR #13268:
URL: https://github.com/apache/kafka/pull/13268#discussion_r1112735419
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -704,18 +708,18 @@ abstract class AbstractFetcherThread(name: String,
* Putting the two cases together, the follower should fetch from the higher one of its replica log end offset
* and the current leader's (local-log-start-offset or) log start offset.
*/
- val (epoch, leaderStartOffset) = if (fetchFromLocalLogStartOffset)
+ val offsetAndEpoch = if (fetchFromLocalLogStartOffset)
leader.fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch) else
leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)
-
+ val leaderStartOffset = offsetAndEpoch.offset
warn(s"Reset fetch offset for partition $topicPartition from $replicaEndOffset to current " +
s"leader's start offset $leaderStartOffset")
val offsetToFetch =
if (leaderStartOffset > replicaEndOffset) {
// Only truncate log when current leader's log start offset (local log start offset if >= 3.4 version incaseof
// OffsetMovedToTieredStorage error) is greater than follower's log end offset.
// truncateAndBuild returns offset value from which it needs to start fetching.
- truncateAndBuild(epoch, leaderStartOffset)
+ truncateAndBuild(offsetAndEpoch.leaderEpoch, leaderStartOffset)
Review Comment:
Minor - maybe the argument of `truncateAndBuild` could be the `offsetAndEpoch` itself?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on a diff in pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "kowshik (via GitHub)" <gi...@apache.org>.
kowshik commented on code in PR #13268:
URL: https://github.com/apache/kafka/pull/13268#discussion_r1113595191
##########
core/src/main/scala/kafka/server/AbstractFetcherThread.scala:
##########
@@ -704,18 +708,18 @@ abstract class AbstractFetcherThread(name: String,
* Putting the two cases together, the follower should fetch from the higher one of its replica log end offset
* and the current leader's (local-log-start-offset or) log start offset.
*/
- val (epoch, leaderStartOffset) = if (fetchFromLocalLogStartOffset)
+ val offsetAndEpoch = if (fetchFromLocalLogStartOffset)
leader.fetchEarliestLocalOffset(topicPartition, currentLeaderEpoch) else
leader.fetchEarliestOffset(topicPartition, currentLeaderEpoch)
-
+ val leaderStartOffset = offsetAndEpoch.offset
warn(s"Reset fetch offset for partition $topicPartition from $replicaEndOffset to current " +
s"leader's start offset $leaderStartOffset")
val offsetToFetch =
if (leaderStartOffset > replicaEndOffset) {
// Only truncate log when current leader's log start offset (local log start offset if >= 3.4 version incaseof
// OffsetMovedToTieredStorage error) is greater than follower's log end offset.
// truncateAndBuild returns offset value from which it needs to start fetching.
- truncateAndBuild(epoch, leaderStartOffset)
+ truncateAndBuild(offsetAndEpoch.leaderEpoch, leaderStartOffset)
Review Comment:
Done. Good point!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "kowshik (via GitHub)" <gi...@apache.org>.
kowshik commented on PR #13268:
URL: https://github.com/apache/kafka/pull/13268#issuecomment-1439126402
@Hangleton Thanks for the review. I have addressed the comment, the PR is ready for another pass.
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] kowshik commented on pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "kowshik (via GitHub)" <gi...@apache.org>.
kowshik commented on PR #13268:
URL: https://github.com/apache/kafka/pull/13268#issuecomment-1434094602
Hi @junrao and @satishd -- Please could you help review this PR?
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] Hangleton commented on pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "Hangleton (via GitHub)" <gi...@apache.org>.
Hangleton commented on PR #13268:
URL: https://github.com/apache/kafka/pull/13268#issuecomment-1443660722
Thanks!
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
[GitHub] [kafka] junrao merged pull request #13268: MINOR: Introduce OffsetAndEpoch in LeaderEndpoint interface return values
Posted by "junrao (via GitHub)" <gi...@apache.org>.
junrao merged PR #13268:
URL: https://github.com/apache/kafka/pull/13268
--
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: jira-unsubscribe@kafka.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org