KAFKA-9360: Fix heartbeat and checkpoint emission can not turn off

[ning2008wisc]

emit.heartbeats.enabled and emit.checkpoints.enabled are supposed to
be the knobs to control if the heartbeat message or checkpoint message
will be sent or not to the topics respectively. In our experiments,
setting them to false will not suspend the activity in their SourceTasks,
e.g. MirrorHeartbeatTask, MirrorCheckpointTask.

The observations are, when setting those knobs to false, huge volume of
SourceRecord are being sent without interval, causing significantly high
CPU usage of MirrorMaker 2 instance, GC time and congesting the single partition of
the heartbeat topic and checkpoint topic.

The proposed fix in the following PR is to (1) explicitly check if interval
is set to negative (e.g. -1), when the emit.heartbeats.enabled or
emit.checkpoints.enabled is off. (2) if interval is indeed set to negative,
put the thread in sleep mode for a while (e.g. 5 seconds) and return null, in order
to prevent it from (1) hogging the cpu, (2) sending heartbeat or checkpoint messages
to Kafka topics

Committer Checklist (excluded from commit message)

• Verify design and implementation
• Verify test coverage and CI build status
• Verify documentation (including upgrade notes)

[ryannedolan]

Thanks for finding this bug! Your approach looks fine, but I think we can do better. You'll see I've got this logic already:

stopped.await(interval.toMillis(), MILLISECONDS)

... which will wait forever unless the task is stopped. As soon as the task is stopped, e.g. during a rebalance, the await() returns and unblocks the WorkerSourceTask.

I think we can do something like that in poll() too. If interval is negative, just await MAX_LONG or something:

if (interval.isNegative()) {
    stopped.await(Long.MAX_VALUE, TimeUnit.MILLISECONDS)
}

[ning2008wisc]

@ryannedolan great comments. updated the PR

[ning2008wisc]

@mimaison when you have some free cycle, do you mind to review this PR? Thanks

[mimaison]

Thanks for the PR.

Is it possible to not even start the connectors if they are disabled? Could we do that in https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java#L203?

[ryannedolan]

Another approach is to create zero tasks when these are disabled. (i.e. create the connectors but no tasks).

[ning2008wisc]

@ryannedolan @mimaison do you think it is a cleaner solution if we could dynamically generate the CONNECTOR_CLASSES https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java#L95-L98 based on the MirrorConnectorConfig https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorConnectorConfig.java#L158 ?

[ryannedolan]

I'm fine with any of these approaches. The nice thing about having the Connectors and/or Tasks run, even if they don't do anything, is they could potentially still emit metrics and log messages, maybe.

You'll notice that Herders are started for each cluster-cluster pair (in MirrorMaker.java), even for pairs that are "disabled". This is because we want to emit Heartbeats even when no replication is done. So you can have A->B.enabled = false and no replication will be done, but you'll still see heartbeats emitted to B (just not replicated from A back to B).

At one point I had logic in MirrorMaker.java to not create herders for disabled flows, as you might expect would happen. But then almost everything broke, because so much depends on heartbeats. For example, if you have A->B enabled and B->A disabled, you'd get no heartbeats in either direction. This is because the A->B herder emits heartbeats to B, and the B->A herder replicates them from B->A. Without both herders (and both MirrorSourceConnectors) running in both directions, you don't see any replicated heartbeats, even tho data is being replicated!

For this reason, MM2 always creates all herders for all cluster-cluster pairs. For flows that are disabled, MirrorSourceConnector just doesn't create any replication tasks.

I suspect we should do the same with heartbeats and checkpoints. If these are disabled, the herder should still get created, the Connectors should still run, but they should just do nothing.

[ning2008wisc]

@ryannedolan thanks for providing your thoughts on the design. I have a couple of questions:

(1) if heartbeat is to detect the liveness of the system which is important with minimum traffic/load, should we still expose the knob to turn off the heartbeat or always make it on?

(2) I may not see how the heartbeat topic will be used or consumed in the kafka open-source code, so does it mean the consumption of the heartbeat topic leave up to the users to implement some healthiness monitor/tool in house?

[ryannedolan]

(1) if heartbeat is to detect the liveness of the system which is important with minimum traffic/load, should we still expose the knob to turn off the heartbeat or always make it on?

Heartbeats are necessary for RemoteClusterUtils to work, e.g. upstreamClusters() just looks for replicated heartbeat topics. However, technically everything works so long as the topics exist -- even if there are no actual heartbeat records. I think, ideally, the MirrorHeartbeatConnector would always run and always create the appropriate topics, even if heartbeats are disabled. Then, emit.heartbeats.enabled=false would just not run the tasks, or run them as no-ops like you've implemented.

(2) I may not see how the heartbeat topic will be used or consumed in the kafka open-source code, so does it mean the consumption of the heartbeat topic leave up to the users to implement some healthiness monitor/tool in house?

Heartbeat messages are not consumed by the open-source code, tho you are correct in assuming that third-party tooling (such as Cloudera's Streams Replication Manager) has been built to use them. But again, the topics are indeed used by RemoteClusterUtils -- just not the records.

It's also worth noting that heartbeats could come from external systems. Anything could emit heartbeats in any format, and the RemoteClusterUtils stuff would still work.

[ning2008wisc]

thanks so much for the details.
@ryannedolan @mimaison Since this still seems a bug, after discussion, could we align on the approach to fix this? For example, MirrorHeartbeatConnector is still created, but no task is created, when emit.heartbeats.enabled=false ?

[ning2008wisc]

@mimaison @ryannedolan see if the updated PR works better, I tested it in my environment.

[ryannedolan]

I like this approach 👌 lgtm, thanks!

[ning2008wisc]

@mimaison any thoughts on this approach? Thanks

[ning2008wisc]

hi @mimaison understand that this is not super critical, but since it is a simple fix, just want to check in and see if you think this approach is acceptable. would like to hear your thoughts and comments.

[mimaison]

Should we check for !config.enabled() here too?

[ryannedolan]

No, we want to emit heartbeats even if A->B.enabled=false (see my earlier comment). Basically, we want heartbeats to be sent even if replication is disabled, unless heartbeats are explicitly disabled with emit.heartbeats.enabled = false.

[mimaison]

Gotcha, maybe worth adding that the explanation to the comment

[ning2008wisc]

Great, I will add the the explanation to the comment in the next update

[ning2008wisc]

if isNegative() is not highly recommended, what about this change? Should we still check isNegative() or return the interval as Duration.ofMillis(Long.MAX_VALUE) ?

[mimaison]

It feels like allowing intervals to be negative interval to disable connectors is not a great pattern. Instead preventing the interval to be negative and relying solely on enabled() would have been a better alternative.

[ning2008wisc]

@ryannedolan , as @mimaison suggested, if emit.checkpoints.enabled = false, do you think if it is a better alternative to return Long.MAX_VALUE, rather than negative?

    Duration emitCheckpointsInterval() {
        if (getBoolean(EMIT_CHECKPOINTS_ENABLED)) {
            return Duration.ofSeconds(getLong(EMIT_CHECKPOINTS_INTERVAL_SECONDS));
        } else {
            // negative interval to disable
            return Duration.ofMillis(Long.MAX_VALUE);
        }
    }

[ryannedolan]

All the interval configs (sync topics acls, sync topic configs, refresh topics, ...) use negative values to indicate that the Scheduler should not run the corresponding task. I suppose we could have used an Optional or something instead. If we drop the negatives here, we should do so with all the other tasks. Mox nix to me.

[mimaison]

It's not ideal but yes we don't want to change the current behaviour here. Let's keep the negative values.

[ning2008wisc]

great, thanks for all of your inputs. Then I suppose the only change in the next update is to adding the explanation to the comment on why we use isNegative()to disable theheartbeatandcheckpoint` emissions.

[ning2008wisc]

@ryannedolan @mimaison updated the comments in the code, please take another review when possible. Thanks

[ryannedolan]

lgtm, thx again!

[mimaison]

Sorry for all the back and forth so such a small change but can you add a couple of small tests that cover this change?

[ning2008wisc]

@mimaison tests are added. Thanks for your great feedback

[mimaison]

@ning2008wisc Thanks for the quick update. The change looks good but it's failing checkstyle as there's a couple of unused imports.
To run checkstyle, run:

gradle
./gradlew connect:mirror:test

[ning2008wisc]

@mimaison thanks for showing how to do checkstyle. Now my local run of checkstyle returns success.

[mimaison]

retest this please

[mimaison]

LGTM