Skip to content

KAFKA-14274: [1/7] basic refactoring #14305

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
junrao merged 1 commit into from
Sep 7, 2023
Merged

KAFKA-14274: [1/7] basic refactoring #14305

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
18 changes: 18 additions & 0 deletions clients/src/main/java/org/apache/kafka/clients/NetworkClientUtils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.kafka.clients;

import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.AuthenticationException;
import org.apache.kafka.common.errors.DisconnectException;
import org.apache.kafka.common.utils.Time;

Expand Down Expand Up @@ -114,4 +115,21 @@ public static ClientResponse sendAndReceive(KafkaClient client, ClientRequest re

}
}

/**
* Check if the code is disconnected and unavailable for immediate reconnection (i.e. if it is in
* reconnect backoff window following the disconnect).
*/
public static boolean isUnavailable(KafkaClient client, Node node, Time time) {
return client.connectionFailed(node) && client.connectionDelay(node, time.milliseconds()) > 0;
}

/**
* Check for an authentication error on a given node and raise the exception if there is one.
*/
public static void maybeThrowAuthFailure(KafkaClient client, Node node) {
AuthenticationException exception = client.authenticationException(node);
if (exception != null)
throw exception;
}
}
6 changes: 3 additions & 3 deletions clients/src/main/java/org/apache/kafka/clients/consumer/ConsumerConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -639,9 +639,9 @@ private void maybeOverrideClientId(Map<String, Object> configs) {
}
}

protected static Map<String, Object> appendDeserializerToConfig(Map<String, Object> configs,
Deserializer<?> keyDeserializer,
Deserializer<?> valueDeserializer) {
public static Map<String, Object> appendDeserializerToConfig(Map<String, Object> configs,

@clolov clolov Sep 7, 2023

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I assume this visibility change is needed for a subsequent commit?

@kirktrue kirktrue Sep 7, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Yes.

Deserializer<?> keyDeserializer,
Deserializer<?> valueDeserializer) {
// validate deserializer configuration, if the passed deserializer instance is null, the user must explicitly set a valid deserializer configuration value
Map<String, Object> newConfigs = new HashMap<>(configs);
if (keyDeserializer != null)
Expand Down
22 changes: 11 additions & 11 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/AbstractFetch.java
View file
Open in desktop

@kirktrue kirktrue Aug 29, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Author’s note: most of these changes are related to the encapsulation changes that were made in CompletedFetch.

Original file line number Diff line number Diff line change
Expand Up @@ -274,11 +274,11 @@ public Fetch<K, V> collectFetch() {

try {
while (recordsRemaining > 0) {
if (nextInLineFetch == null || nextInLineFetch.isConsumed) {
if (nextInLineFetch == null || nextInLineFetch.isConsumed()) {
CompletedFetch<K, V> records = completedFetches.peek();
if (records == null) break;

if (!records.initialized) {
if (!records.isInitialized()) {
try {
nextInLineFetch = initializeCompletedFetch(records);
} catch (Exception e) {
Expand Down Expand Up @@ -336,18 +336,18 @@ private Fetch<K, V> fetchRecords(final int maxRecords) {
throw new IllegalStateException("Missing position for fetchable partition " + nextInLineFetch.partition);
}

if (nextInLineFetch.nextFetchOffset == position.offset) {
if (nextInLineFetch.nextFetchOffset() == position.offset) {
List<ConsumerRecord<K, V>> partRecords = nextInLineFetch.fetchRecords(maxRecords);

log.trace("Returning {} fetched records at offset {} for assigned partition {}",
partRecords.size(), position, nextInLineFetch.partition);

boolean positionAdvanced = false;

if (nextInLineFetch.nextFetchOffset > position.offset) {
if (nextInLineFetch.nextFetchOffset() > position.offset) {
SubscriptionState.FetchPosition nextPosition = new SubscriptionState.FetchPosition(
nextInLineFetch.nextFetchOffset,
nextInLineFetch.lastEpoch,
nextInLineFetch.nextFetchOffset(),
nextInLineFetch.lastEpoch(),
position.currentLeader);
log.trace("Updating fetch position from {} to {} for partition {} and returning {} records from `poll()`",
position, nextPosition, nextInLineFetch.partition, partRecords.size());
Expand All @@ -369,7 +369,7 @@ private Fetch<K, V> fetchRecords(final int maxRecords) {
// these records aren't next in line based on the last consumed position, ignore them
// they must be from an obsolete request
log.debug("Ignoring fetched records for {} at offset {} since the current position is {}",
nextInLineFetch.partition, nextInLineFetch.nextFetchOffset, position);
nextInLineFetch.partition, nextInLineFetch.nextFetchOffset(), position);
}
}

Expand All @@ -381,7 +381,7 @@ private Fetch<K, V> fetchRecords(final int maxRecords) {

private List<TopicPartition> fetchablePartitions() {
Set<TopicPartition> exclude = new HashSet<>();
if (nextInLineFetch != null && !nextInLineFetch.isConsumed) {
if (nextInLineFetch != null && !nextInLineFetch.isConsumed()) {
exclude.add(nextInLineFetch.partition);
}
for (CompletedFetch<K, V> completedFetch : completedFetches) {
Expand Down Expand Up @@ -528,7 +528,7 @@ private CompletedFetch<K, V> initializeCompletedFetch(final CompletedFetch<K, V>

private CompletedFetch<K, V> handleInitializeCompletedFetchSuccess(final CompletedFetch<K, V> completedFetch) {
final TopicPartition tp = completedFetch.partition;
final long fetchOffset = completedFetch.nextFetchOffset;
final long fetchOffset = completedFetch.nextFetchOffset();

// we are interested in this fetch only if the beginning offset matches the
// current consumed position
Expand Down Expand Up @@ -586,14 +586,14 @@ private CompletedFetch<K, V> handleInitializeCompletedFetchSuccess(final Complet
});
}

completedFetch.initialized = true;
completedFetch.setInitialized();
return completedFetch;
}

private void handleInitializeCompletedFetchErrors(final CompletedFetch<K, V> completedFetch,
final Errors error) {
final TopicPartition tp = completedFetch.partition;
final long fetchOffset = completedFetch.nextFetchOffset;
final long fetchOffset = completedFetch.nextFetchOffset();

if (error == Errors.NOT_LEADER_OR_FOLLOWER ||
error == Errors.REPLICA_NOT_AVAILABLE ||
Expand Down
26 changes: 19 additions & 7 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/CommitRequestManager.java
View file
Open in desktop

@kirktrue kirktrue Aug 29, 2023

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Author’s note: renamed subscriptionState to subscriptions for consistency with the rest of the newly refactored code base.

Original file line number Diff line number Diff line change
Expand Up @@ -53,7 +53,8 @@ public class CommitRequestManager implements RequestManager {
// TODO: current in ConsumerConfig but inaccessible in the internal package.
private static final String THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED = "internal.throw.on.fetch.stable.offset.unsupported";
// TODO: We will need to refactor the subscriptionState
private final SubscriptionState subscriptionState;
private final SubscriptionState subscriptions;
private final LogContext logContext;
private final Logger log;
private final Optional<AutoCommitState> autoCommitState;
private final CoordinatorRequestManager coordinatorRequestManager;
Expand All @@ -66,11 +67,12 @@ public class CommitRequestManager implements RequestManager {
public CommitRequestManager(
final Time time,
final LogContext logContext,
final SubscriptionState subscriptionState,
final SubscriptionState subscriptions,
final ConsumerConfig config,
final CoordinatorRequestManager coordinatorRequestManager,
final GroupState groupState) {
Objects.requireNonNull(coordinatorRequestManager, "Coordinator is needed upon committing offsets");
this.logContext = logContext;
this.log = logContext.logger(getClass());
this.pendingRequests = new PendingRequests();
if (config.getBoolean(ConsumerConfig.ENABLE_AUTO_COMMIT_CONFIG)) {
Expand All @@ -82,7 +84,7 @@ public CommitRequestManager(
}
this.coordinatorRequestManager = coordinatorRequestManager;
this.groupState = groupState;
this.subscriptionState = subscriptionState;
this.subscriptions = subscriptions;
this.retryBackoffMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MS_CONFIG);
this.retryBackoffMaxMs = config.getLong(ConsumerConfig.RETRY_BACKOFF_MAX_MS_CONFIG);
this.throwOnFetchStableOffsetUnsupported = config.getBoolean(THROW_ON_FETCH_STABLE_OFFSET_UNSUPPORTED);
Expand All @@ -99,7 +101,7 @@ public NetworkClientDelegate.PollResult poll(final long currentTimeMs) {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList());
}

maybeAutoCommit(this.subscriptionState.allConsumed());
maybeAutoCommit(this.subscriptions.allConsumed());
if (!pendingRequests.hasUnsentRequests()) {
return new NetworkClientDelegate.PollResult(Long.MAX_VALUE, Collections.emptyList());
}
Expand Down Expand Up @@ -167,9 +169,9 @@ CompletableFuture<ClientResponse> sendAutoCommit(final Map<TopicPartition, Offse
})
.exceptionally(t -> {
if (t instanceof RetriableCommitFailedException) {
log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, t);
log.debug("Asynchronous auto-commit of offsets {} failed due to retriable error: {}", allConsumedOffsets, t.getMessage());
} else {
log.warn("Asynchronous auto-commit of offsets {} failed: {}", allConsumedOffsets, t.getMessage());
log.warn("Asynchronous auto-commit of offsets {} failed", allConsumedOffsets, t);
}
return null;
});
Expand Down Expand Up @@ -241,7 +243,7 @@ public OffsetFetchRequestState(final Set<TopicPartition> partitions,
final GroupState.Generation generation,
final long retryBackoffMs,
final long retryBackoffMaxMs) {
super(retryBackoffMs, retryBackoffMaxMs);
super(logContext, CommitRequestManager.class.getSimpleName(), retryBackoffMs, retryBackoffMaxMs);
this.requestedPartitions = partitions;
this.requestedGeneration = generation;
this.future = new CompletableFuture<>();
Expand Down Expand Up @@ -366,6 +368,16 @@ private CompletableFuture<Map<TopicPartition, OffsetAndMetadata>> chainFuture(fi
}
});
}

@Override
public String toString() {
return "OffsetFetchRequestState{" +
"requestedPartitions=" + requestedPartitions +
", requestedGeneration=" + requestedGeneration +
", future=" + future +
", " + toStringBase() +
'}';
}
}

/**
Expand Down
32 changes: 26 additions & 6 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/CompletedFetch.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -57,17 +57,12 @@
* @param <K> Record key type
* @param <V> Record value type
*/
class CompletedFetch<K, V> {
public class CompletedFetch<K, V> {

final TopicPartition partition;
final FetchResponseData.PartitionData partitionData;
final short requestVersion;

long nextFetchOffset;
Optional<Integer> lastEpoch;
boolean isConsumed = false;
boolean initialized = false;

private final Logger log;
private final SubscriptionState subscriptions;
private final FetchConfig<K, V> fetchConfig;
Expand All @@ -84,6 +79,10 @@ class CompletedFetch<K, V> {
private CloseableIterator<Record> records;
private Exception cachedRecordException = null;
private boolean corruptLastRecord = false;
private long nextFetchOffset;
private Optional<Integer> lastEpoch;
private boolean isConsumed = false;
private boolean initialized = false;

CompletedFetch(LogContext logContext,
SubscriptionState subscriptions,
Expand All @@ -109,6 +108,27 @@ class CompletedFetch<K, V> {
this.abortedTransactions = abortedTransactions(partitionData);
}

long nextFetchOffset() {
return nextFetchOffset;
}

Optional<Integer> lastEpoch() {
return lastEpoch;
}

boolean isInitialized() {
return initialized;
}

void setInitialized() {
this.initialized = true;
}

public boolean isConsumed() {
return isConsumed;
}


/**
* After each partition is parsed, we update the current metric totals with the total bytes
* and number of records parsed. After all partitions have reported, we write the metric.
Expand Down
7 changes: 3 additions & 4 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerNetworkClient.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,7 @@
import org.apache.kafka.clients.ClientResponse;
import org.apache.kafka.clients.KafkaClient;
import org.apache.kafka.clients.Metadata;
import org.apache.kafka.clients.NetworkClientUtils;
import org.apache.kafka.clients.RequestCompletionHandler;
import org.apache.kafka.common.Node;
import org.apache.kafka.common.errors.AuthenticationException;
Expand Down Expand Up @@ -558,7 +559,7 @@ public void close() throws IOException {
public boolean isUnavailable(Node node) {
lock.lock();
try {
return client.connectionFailed(node) && client.connectionDelay(node, time.milliseconds()) > 0;
return NetworkClientUtils.isUnavailable(client, node, time);
} finally {
lock.unlock();
}
Expand All @@ -570,9 +571,7 @@ public boolean isUnavailable(Node node) {
public void maybeThrowAuthFailure(Node node) {
lock.lock();
try {
AuthenticationException exception = client.authenticationException(node);
if (exception != null)
throw exception;
NetworkClientUtils.maybeThrowAuthFailure(client, node);
} finally {
lock.unlock();
}
Expand Down
12 changes: 8 additions & 4 deletions .../src/main/java/org/apache/kafka/clients/consumer/internals/CoordinatorRequestManager.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -34,13 +34,13 @@

/**
* This is responsible for timing to send the next {@link FindCoordinatorRequest} based on the following criteria:
*
* <p/>
* Whether there is an existing coordinator.
* Whether there is an inflight request.
* Whether the backoff timer has expired.
* The {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.PollResult} contains either a wait timer
* or a singleton list of {@link org.apache.kafka.clients.consumer.internals.NetworkClientDelegate.UnsentRequest}.
*
* <p/>
* The {@link FindCoordinatorRequest} will be handled by the {@link #onResponse(long, FindCoordinatorResponse)} callback, which
* subsequently invokes {@code onResponse} to handle the exception and response. Note that the coordinator node will be
* marked {@code null} upon receiving a failure.
Expand Down Expand Up @@ -70,7 +70,12 @@ public CoordinatorRequestManager(
this.log = logContext.logger(this.getClass());
this.nonRetriableErrorHandler = errorHandler;
this.groupId = groupId;
this.coordinatorRequestState = new RequestState(retryBackoffMs, retryBackoffMaxMs);
this.coordinatorRequestState = new RequestState(
logContext,
CoordinatorRequestManager.class.getSimpleName(),
retryBackoffMs,
retryBackoffMaxMs
);
}

/**
Expand Down Expand Up @@ -218,5 +223,4 @@ private void onResponse(
public Optional<Node> coordinator() {
return Optional.ofNullable(this.coordinator);
}

}
Loading