Repository: apache/kafka
File: clients/src/main/java/org/apache/kafka/clients/consumer/internals/SubscriptionState.java (Java)

Code before:
        TopicPartitionState topicPartitionState = assignedState(tp);
        if (isolationLevel == IsolationLevel.READ_COMMITTED) {
            return topicPartitionState.lastStableOffset;
        } else {
            return topicPartitionState.highWatermark;
        }
    }

    public synchronized void requestPartitionEndOffset(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedState(tp);
        topicPartitionState.setRequestEndOffset(true);
    }

    public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedState(tp);
        return topicPartitionState.endOffsetRequested();
    }

    public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedStateOrNull(tp);

        if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
            topicPartitionState.setRequestEndOffset(false);
            return true;
        } else {
            return false;
        }
    }

    synchronized Long partitionLead(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedState(tp);
        return topicPartitionState.logStartOffset == null ? null : topicPartitionState.position.offset - topicPartitionState.logStartOffset;
    }

    synchronized void updateHighWatermark(TopicPartition tp, long highWatermark) {
        assignedState(tp).highWatermark(highWatermark);
    }

    synchronized boolean tryUpdatingHighWatermark(TopicPartition tp, long highWatermark) {
        final TopicPartitionState state = assignedStateOrNull(tp);
        if (state != null) {
            assignedState(tp).highWatermark(highWatermark);
            return true;
        }
        return false;
    }

    synchronized boolean tryUpdatingLogStartOffset(TopicPartition tp, long highWatermark) {
        final TopicPartitionState state = assignedStateOrNull(tp);
        if (state != null) {
            assignedState(tp).logStartOffset(highWatermark);

Review comment (refactor):
this return type seems to only be used to log if we performed the action or not. Should we move the log here instead? (simplify this func and the caller) 

Code after:
        TopicPartitionState topicPartitionState = assignedState(tp);
        if (isolationLevel == IsolationLevel.READ_COMMITTED) {
            return topicPartitionState.lastStableOffset;
        } else {
            return topicPartitionState.highWatermark;
        }
    }

    public synchronized void requestPartitionEndOffset(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedState(tp);
        topicPartitionState.requestEndOffset();
    }

    public synchronized boolean partitionEndOffsetRequested(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedState(tp);
        return topicPartitionState.endOffsetRequested();
    }

    public synchronized boolean maybeClearPartitionEndOffsetRequested(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedStateOrNull(tp);

        if (topicPartitionState != null && topicPartitionState.endOffsetRequested()) {
            topicPartitionState.clearEndOffset();
            return true;
        } else {
            return false;
        }
    }

    synchronized Long partitionLead(TopicPartition tp) {
        TopicPartitionState topicPartitionState = assignedState(tp);
        return topicPartitionState.logStartOffset == null ? null : topicPartitionState.position.offset - topicPartitionState.logStartOffset;
    }

    synchronized void updateHighWatermark(TopicPartition tp, long highWatermark) {
        assignedState(tp).highWatermark(highWatermark);
    }

    synchronized boolean tryUpdatingHighWatermark(TopicPartition tp, long highWatermark) {
        final TopicPartitionState state = assignedStateOrNull(tp);
        if (state != null) {
            assignedState(tp).highWatermark(highWatermark);
            return true;
        }
        return false;
    }

    synchronized boolean tryUpdatingLogStartOffset(TopicPartition tp, long highWatermark) {
        final TopicPartitionState state = assignedStateOrNull(tp);
        if (state != null) {
            assignedState(tp).logStartOffset(highWatermark);