Skip to content

KAFKA-8449: Restart tasks on reconfiguration under incremental cooperative rebalancing #6850

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
rhauch merged 3 commits into from
Jun 3, 2019
Merged

KAFKA-8449: Restart tasks on reconfiguration under incremental cooperative rebalancing #6850

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
1669f4d
KAFKA-8449: Restart task on reconfiguration under incremental coopera…
kkonstantine May 31, 2019
f42e3af
KAFKA-8449: Keep execution paths separate for config updates between …
kkonstantine Jun 3, 2019
f3383ca
KAFKA-8449: Add comments and improve logging
kkonstantine Jun 3, 2019
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
203 changes: 159 additions & 44 deletions ...runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -80,6 +80,8 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.stream.Collectors;

import static org.apache.kafka.connect.runtime.distributed.ConnectProtocol.CONNECT_PROTOCOL_V0;

Expand Down Expand Up @@ -141,6 +143,7 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
// and the from other nodes are safe to process
private boolean rebalanceResolved;
private ExtendedAssignment runningAssignment = ExtendedAssignment.empty();
private Set<ConnectorTaskId> tasksToRestart = new HashSet<>();
private ExtendedAssignment assignment;
private boolean canReadConfigs;
private ClusterConfigState configState;
Expand All @@ -151,6 +154,7 @@ public class DistributedHerder extends AbstractHerder implements Runnable {
// Config updates can be collected and applied together when possible. Also, we need to take care to rebalance when
// needed (e.g. task reconfiguration, which requires everyone to coordinate offset commits).
private Set<String> connectorConfigUpdates = new HashSet<>();
private Set<ConnectorTaskId> taskConfigUpdates = new HashSet<>();
// Similarly collect target state changes (when observed by the config storage listener) for handling in the
// herder's main thread.
private Set<String> connectorTargetStateChanges = new HashSet<>();
Expand Down Expand Up @@ -304,51 +308,47 @@ public void tick() {
}

// Process any configuration updates
Set<String> connectorConfigUpdatesCopy = null;
Set<String> connectorTargetStateChangesCopy = null;
synchronized (this) {
if (needsReconfigRebalance || !connectorConfigUpdates.isEmpty() || !connectorTargetStateChanges.isEmpty()) {
// Connector reconfigs only need local updates since there is no coordination between workers required.
// However, if connectors were added or removed, work needs to be rebalanced since we have more work
// items to distribute among workers.
configState = configBackingStore.snapshot();

if (needsReconfigRebalance) {
// Task reconfigs require a rebalance. Request the rebalance, clean out state, and then restart
// this loop, which will then ensure the rebalance occurs without any other requests being
// processed until it completes.
member.requestRejoin();
// Any connector config updates or target state changes will be addressed during the rebalance too
connectorConfigUpdates.clear();
connectorTargetStateChanges.clear();
needsReconfigRebalance = false;
log.debug("Requesting rebalance due to reconfiguration of tasks (needsReconfigRebalance: {})",
needsReconfigRebalance);
return;
} else {
if (!connectorConfigUpdates.isEmpty()) {
// We can't start/stop while locked since starting connectors can cause task updates that will
// require writing configs, which in turn make callbacks into this class from another thread that
// require acquiring a lock. This leads to deadlock. Instead, just copy the info we need and process
// the updates after unlocking.
connectorConfigUpdatesCopy = connectorConfigUpdates;
connectorConfigUpdates = new HashSet<>();
}
AtomicReference<Set<String>> connectorConfigUpdatesCopy = new AtomicReference<>();
AtomicReference<Set<String>> connectorTargetStateChangesCopy = new AtomicReference<>();
AtomicReference<Set<ConnectorTaskId>> taskConfigUpdatesCopy = new AtomicReference<>();

boolean shouldReturn;
if (member.currentProtocolVersion() == CONNECT_PROTOCOL_V0) {
shouldReturn = updateConfigsWithEager(connectorConfigUpdatesCopy,
connectorTargetStateChangesCopy);
// With eager protocol we should return immediately if needsReconfigRebalance has
// been set to retain the old workflow
if (shouldReturn) {
return;
}

if (!connectorTargetStateChanges.isEmpty()) {
// Similarly for target state changes which can cause connectors to be restarted
connectorTargetStateChangesCopy = connectorTargetStateChanges;
connectorTargetStateChanges = new HashSet<>();
}
}
if (connectorConfigUpdatesCopy.get() != null) {
processConnectorConfigUpdates(connectorConfigUpdatesCopy.get());
}
}

if (connectorConfigUpdatesCopy != null)
processConnectorConfigUpdates(connectorConfigUpdatesCopy);
if (connectorTargetStateChangesCopy.get() != null) {
processTargetStateChanges(connectorTargetStateChangesCopy.get());
}
} else {
shouldReturn = updateConfigsWithIncrementalCooperative(connectorConfigUpdatesCopy,
connectorTargetStateChangesCopy, taskConfigUpdatesCopy);

if (connectorConfigUpdatesCopy.get() != null) {
processConnectorConfigUpdates(connectorConfigUpdatesCopy.get());
}

if (connectorTargetStateChangesCopy.get() != null) {
processTargetStateChanges(connectorTargetStateChangesCopy.get());
}

if (connectorTargetStateChangesCopy != null)
processTargetStateChanges(connectorTargetStateChangesCopy);
if (taskConfigUpdatesCopy.get() != null) {
processTaskConfigUpdatesWithIncrementalCooperative(taskConfigUpdatesCopy.get());
}

if (shouldReturn) {
return;
}
}

// Let the group take any actions it needs to
try {
Expand All @@ -360,6 +360,95 @@ public void tick() {
}
}

private synchronized boolean updateConfigsWithEager(AtomicReference<Set<String>> connectorConfigUpdatesCopy,
AtomicReference<Set<String>> connectorTargetStateChangesCopy) {
// This branch is here to avoid creating a snapshot if not needed
if (needsReconfigRebalance
|| !connectorConfigUpdates.isEmpty()
|| !connectorTargetStateChanges.isEmpty()) {
// Connector reconfigs only need local updates since there is no coordination between workers required.
// However, if connectors were added or removed, work needs to be rebalanced since we have more work
// items to distribute among workers.
configState = configBackingStore.snapshot();

if (needsReconfigRebalance) {
// Task reconfigs require a rebalance. Request the rebalance, clean out state, and then restart
// this loop, which will then ensure the rebalance occurs without any other requests being
// processed until it completes.
log.debug("Requesting rebalance due to reconfiguration of tasks (needsReconfigRebalance: {})",
needsReconfigRebalance);
member.requestRejoin();
needsReconfigRebalance = false;
// Any connector config updates or target state changes will be addressed during the rebalance too
connectorConfigUpdates.clear();
connectorTargetStateChanges.clear();
return true;
} else {
if (!connectorConfigUpdates.isEmpty()) {
// We can't start/stop while locked since starting connectors can cause task updates that will
// require writing configs, which in turn make callbacks into this class from another thread that
// require acquiring a lock. This leads to deadlock. Instead, just copy the info we need and process
// the updates after unlocking.
connectorConfigUpdatesCopy.set(connectorConfigUpdates);
connectorConfigUpdates = new HashSet<>();
}

if (!connectorTargetStateChanges.isEmpty()) {
// Similarly for target state changes which can cause connectors to be restarted
connectorTargetStateChangesCopy.set(connectorTargetStateChanges);
connectorTargetStateChanges = new HashSet<>();
}
}
}
return false;
}

private synchronized boolean updateConfigsWithIncrementalCooperative(AtomicReference<Set<String>> connectorConfigUpdatesCopy,
AtomicReference<Set<String>> connectorTargetStateChangesCopy,
AtomicReference<Set<ConnectorTaskId>> taskConfigUpdatesCopy) {
boolean retValue = false;
// This branch is here to avoid creating a snapshot if not needed
if (needsReconfigRebalance
|| !connectorConfigUpdates.isEmpty()
|| !connectorTargetStateChanges.isEmpty()
|| !taskConfigUpdates.isEmpty()) {
// Connector reconfigs only need local updates since there is no coordination between workers required.
// However, if connectors were added or removed, work needs to be rebalanced since we have more work
// items to distribute among workers.
configState = configBackingStore.snapshot();

if (needsReconfigRebalance) {
log.debug("Requesting rebalance due to reconfiguration of tasks (needsReconfigRebalance: {})",

@rhauch rhauch Jun 2, 2019

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.

Minor: this log line is in a different location in the similar updateConfigsWithEager(...) method. Would it make sense to move this log message to just before the member.requestRejoin() call?

@kkonstantine kkonstantine Jun 3, 2019

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.

Moved it to both before member.requestRejoin. I changed it because it was even after we set this variable to false, which made the log message a bit confusing.

@kkonstantine kkonstantine Jun 3, 2019

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.

(done)

needsReconfigRebalance);
member.requestRejoin();
needsReconfigRebalance = false;
retValue = true;
}

if (!connectorConfigUpdates.isEmpty()) {
// We can't start/stop while locked since starting connectors can cause task updates that will
// require writing configs, which in turn make callbacks into this class from another thread that
// require acquiring a lock. This leads to deadlock. Instead, just copy the info we need and process
// the updates after unlocking.
connectorConfigUpdatesCopy.set(connectorConfigUpdates);
connectorConfigUpdates = new HashSet<>();
}

if (!connectorTargetStateChanges.isEmpty()) {
// Similarly for target state changes which can cause connectors to be restarted
connectorTargetStateChangesCopy.set(connectorTargetStateChanges);
connectorTargetStateChanges = new HashSet<>();
}

if (!taskConfigUpdates.isEmpty()) {
// Similarly for task config updates
taskConfigUpdatesCopy.set(taskConfigUpdates);
taskConfigUpdates = new HashSet<>();
}
}
return retValue;
}

private void processConnectorConfigUpdates(Set<String> connectorConfigUpdates) {
// If we only have connector config updates, we can just bounce the updated connectors that are
// currently assigned to this worker.
Expand Down Expand Up @@ -396,6 +485,21 @@ private void processTargetStateChanges(Set<String> connectorTargetStateChanges)
}
}

private void processTaskConfigUpdatesWithIncrementalCooperative(Set<ConnectorTaskId> taskConfigUpdates) {
Set<ConnectorTaskId> localTasks = assignment == null
? Collections.emptySet()
: new HashSet<>(assignment.tasks());
Set<String> connectorsWhoseTasksToStop = taskConfigUpdates.stream()
.map(ConnectorTaskId::connector).collect(Collectors.toSet());

List<ConnectorTaskId> tasksToStop = localTasks.stream()
.filter(taskId -> connectorsWhoseTasksToStop.contains(taskId.connector()))
.collect(Collectors.toList());
log.info("Handling task config update by restarting tasks {}", tasksToStop);
worker.stopAndAwaitTasks(tasksToStop);
tasksToRestart.addAll(tasksToStop);
}

// public for testing
public void halt() {
synchronized (this) {
Expand Down Expand Up @@ -900,6 +1004,12 @@ private void startWork() {
callables.add(getConnectorStartingCallable(connectorName));
}

// These tasks have been stopped by this worker due to task reconfiguration. In order to
// restart them, they are removed just before the overall task startup from the set of
// currently running tasks. Therefore, they'll be restarted only if they are included in
// the assignment that was just received after rebalancing.
runningAssignment.tasks().removeAll(tasksToRestart);
tasksToRestart.clear();

@rhauch rhauch Jun 2, 2019

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.

Minor: it would help to have a comment here to explain why the tasks to restart are removed from the running tasks.

@kkonstantine kkonstantine Jun 3, 2019

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.

Done

for (ConnectorTaskId taskId : assignmentDifference(assignment.tasks(), runningAssignment.tasks())) {
callables.add(getTaskStartingCallable(taskId));
}
Expand Down Expand Up @@ -1172,12 +1282,17 @@ public void onConnectorConfigUpdate(String connector) {
public void onTaskConfigUpdate(Collection<ConnectorTaskId> tasks) {
log.info("Tasks {} configs updated", tasks);

// Stage the update and wake up the work thread. No need to record the set of tasks here because task reconfigs
// always need a rebalance to ensure offsets get committed.
// Stage the update and wake up the work thread.
// The set of tasks is recorder for incremental cooperative rebalancing, in which
// tasks don't get restarted unless they are balanced between workers.
// With eager rebalancing there's no need to record the set of tasks because task reconfigs
// always need a rebalance to ensure offsets get committed. In eager rebalancing the
// recorded set of tasks remains unused.
// TODO: As an optimization, some task config updates could avoid a rebalance. In particular, single-task
// connectors clearly don't need any coordination.
synchronized (DistributedHerder.this) {
needsReconfigRebalance = true;
taskConfigUpdates.addAll(tasks);
}
member.wakeup();
}
Expand Down Expand Up @@ -1279,7 +1394,7 @@ public void onAssigned(ExtendedAssignment assignment, int generation) {
// catch up (or backoff if we fail) not executed in a callback, and so we'll be able to invoke other
// group membership actions (e.g., we may need to explicitly leave the group if we cannot handle the
// assigned tasks).
log.info("Joined group and got assignment: {}", assignment);
log.info("Joined group at generation {} and got assignment: {}", generation, assignment);
synchronized (DistributedHerder.this) {
DistributedHerder.this.assignment = assignment;
DistributedHerder.this.generation = generation;
Expand Down
5 changes: 3 additions & 2 deletions ...untime/src/test/java/org/apache/kafka/connect/integration/MonitorableSourceConnector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
public class MonitorableSourceConnector extends TestSourceConnector {
private static final Logger log = LoggerFactory.getLogger(MonitorableSourceConnector.class);

public static final String TOPIC_CONFIG = "topic";
private String connectorName;
private ConnectorHandle connectorHandle;
private Map<String, String> commonConfigs;
Expand Down Expand Up @@ -105,15 +106,15 @@ public String version() {
public void start(Map<String, String> props) {
taskId = props.get("task.id");
connectorName = props.get("connector.name");
topicName = props.getOrDefault("topic", "sequential-topic");
topicName = props.getOrDefault(TOPIC_CONFIG, "sequential-topic");
throughput = Long.valueOf(props.getOrDefault("throughput", "-1"));
batchSize = Integer.valueOf(props.getOrDefault("messages.per.poll", "1"));
taskHandle = RuntimeHandles.get().connectorHandle(connectorName).taskHandle(taskId);
Map<String, Object> offset = Optional.ofNullable(
context.offsetStorageReader().offset(Collections.singletonMap("task.id", taskId)))
.orElse(Collections.emptyMap());
startingSeqno = Optional.ofNullable((Long) offset.get("saved")).orElse(0L);
log.info("Started {} task {}", this.getClass().getSimpleName(), taskId);
log.info("Started {} task {} with properties {}", this.getClass().getSimpleName(), taskId, props);
throttler = new ThroughputThrottler(throughput, System.currentTimeMillis());
}

Expand Down
Loading