Skip to content

KAFKA-9113: P5, StandbyTask Life Cycle #7

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
guozhangwang merged 13 commits into from
Jan 24, 2020
Merged

KAFKA-9113: P5, StandbyTask Life Cycle #7

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
13 commits
Select commit Hold shift + click to select a range
159c48b
added states for tasks and factored out cluster metadata from taskMan…
Jan 10, 2020
34b195c
wip
Jan 12, 2020
66cd779
rebase from John's branch
guozhangwang Jan 17, 2020
cf4eb57
resolve conflicts
guozhangwang Jan 17, 2020
401d169
refactor standby tasks
guozhangwang Jan 17, 2020
58f8a2a
wip
Jan 17, 2020
8e2102c
oof. a lot of work in progress :/
Jan 17, 2020
6b8e86c
wip
Jan 18, 2020
8d8048a
unit tests phase one
guozhangwang Jan 19, 2020
13ce59b
rebase from John's branch
guozhangwang Jan 20, 2020
ef84e91
fixing tests
Jan 21, 2020
be24811
rebase on John's branch
guozhangwang Jan 21, 2020
1c52f56
cleanup StandbyTaskTest
guozhangwang Jan 21, 2020
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
6 changes: 3 additions & 3 deletions clients/src/main/java/org/apache/kafka/clients/consumer/internals/ConsumerCoordinator.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -19,16 +19,16 @@
import org.apache.kafka.clients.GroupRebalanceConfig;
import org.apache.kafka.clients.consumer.CommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.ConsumerGroupMetadata;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Assignment;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.GroupSubscription;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.RebalanceProtocol;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Subscription;
import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
import org.apache.kafka.clients.consumer.OffsetAndMetadata;
import org.apache.kafka.clients.consumer.OffsetCommitCallback;
import org.apache.kafka.clients.consumer.RetriableCommitFailedException;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.Assignment;
import org.apache.kafka.clients.consumer.ConsumerPartitionAssignor.RebalanceProtocol;
import org.apache.kafka.common.Cluster;
import org.apache.kafka.common.KafkaException;
import org.apache.kafka.common.Node;
Expand Down
11 changes: 11 additions & 0 deletions clients/src/main/java/org/apache/kafka/common/utils/Utils.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -842,6 +842,17 @@ public static void closeQuietly(AutoCloseable closeable, String name, AtomicRefe
}
}

public static void executeQuietly(AutoCloseable closeable, String name, AtomicReference<Throwable> firstException) {
if (closeable != null) {
try {
closeable.close();
} catch (Throwable t) {
firstException.compareAndSet(null, t);
log.error("Failed to close {} with type {}", name, closeable.getClass().getName(), t);
}
}
}

/**
* A cheap way to deterministically convert a number to a positive value. When the input is
* positive, the original value is returned. When the input number is negative, the returned
Expand Down
1 change: 1 addition & 0 deletions streams/src/main/java/org/apache/kafka/streams/StreamsConfig.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -796,6 +796,7 @@ public class StreamsConfig extends AbstractConfig {

public static class InternalConfig {
public static final String TASK_MANAGER_FOR_PARTITION_ASSIGNOR = "__task.manager.instance__";
public static final String STREAMS_METADATA_STATE_FOR_PARTITION_ASSIGNOR = "__streams.metadata.state.instance__";
public static final String ASSIGNMENT_ERROR_CODE = "__assignment.error.code__";
}

Expand Down
9 changes: 4 additions & 5 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/AbstractTask.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,6 @@ public abstract class AbstractTask implements Task {
final StateDirectory stateDirectory;

boolean taskInitialized;
boolean taskClosed;
boolean commitNeeded;

InternalProcessorContext processorContext;
Expand Down Expand Up @@ -193,18 +192,18 @@ void registerStateStores() {
store.init(processorContext, store);
log.trace("Registered state store {}", store.name());
}
stateMgr.initStoresFromCheckpointedOffsets();
stateMgr.initializeStoreOffsetsFromCheckpoint();
log.debug("Initialized state stores");
}

/**
* @throws ProcessorStateException if there is an error while closing the state manager
*/
void closeStateManager(final boolean clean) throws ProcessorStateException {
void closeStateManager() throws ProcessorStateException {
ProcessorStateException exception = null;
log.trace("Closing state manager");
try {
stateMgr.close(clean);
stateMgr.close();
} catch (final ProcessorStateException e) {
exception = e;
} finally {
Expand All @@ -222,7 +221,7 @@ void closeStateManager(final boolean clean) throws ProcessorStateException {
}

public boolean isClosed() {
return taskClosed;
return state() == State.CLOSED;
}

public boolean commitNeeded() {
Expand Down
79 changes: 0 additions & 79 deletions streams/src/main/java/org/apache/kafka/streams/processor/internals/AssignedStandbyTasks.java
View file
Open in desktop

This file was deleted.

Loading