[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17556632#comment-17556632
]
Justinwins commented on KAFKA-9981:
---
[~durban] [~vaibhavjaimini] any workaround available for this issue ? i met
this in production env too, and got stuck for a few days.
thanks.
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
> Attachments: image-2022-06-21-12-06-44-010.png
>
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17525490#comment-17525490
]
YANGLiiN commented on KAFKA-9981:
-
the root cause is the connect doesn't process the mm2 add the 'NOTUSED' url on
the slave node
see
[https://github.com/apache/kafka/blob/trunk/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorMaker.java#L122]
we should add the "NOTUSED" logic below the code
[https://github.com/apache/kafka/blob/trunk/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1593-L1599]
{code:java}
if (leaderUrl.startsWith("NOTUSED")) {
configBackingStore.putTaskConfigs(connName, rawTaskProps);
cb.onCompletion(null, null);
return;
}
{code}
PR: https://github.com/apache/kafka/pull/12016
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.20.7#820007)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17393101#comment-17393101
]
Daniel Urban commented on KAFKA-9981:
-
This KIP aims to fix the issue by adding the REST API to MM2, and also
improving the config provider reference handling in the MM2 configs:
[https://cwiki.apache.org/confluence/display/KAFKA/KIP-710%3A+Full+support+for+distributed+mode+in+dedicated+MirrorMaker+2.0+clusters]
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17212107#comment-17212107
]
victor commented on KAFKA-9981:
---
[~vaibhavjaimini] That's how My prd environment works,I also do dynamic
whitelist based on ZooKeeper
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17209466#comment-17209466
]
Vaibhav Jaimini commented on KAFKA-9981:
[~qq619618919] I also faced the same issue with our multiple node cross DC
mirror maker. I've tested the changes that you've made and they are working.
Can you confirm if you went ahead with these or you found another workaround ?
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17113787#comment-17113787
]
Chris Egerton commented on KAFKA-9981:
--
[~qq619618919] I'm afraid I don't know too much about MirrorMaker 2 itself; I'm
more familiar with the Connect framework that it's built on top of.
[~ryannedolan] may know more?
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17112690#comment-17112690
]
victor commented on KAFKA-9981:
---
[~ChrisEgerton] mm2 has realized data backup. How does kafkaproduce realize
automatic failover transparently?
How can the same kafkaproduce object automatically switch between two clusters?
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17110488#comment-17110488
]
Chris Egerton commented on KAFKA-9981:
--
[~qq619618919] we could but it wouldn't be simple. We'd have to take care to
ensure that writes from zombie workers would be ignored, which is done right
now by only allowing the leader to write to the config topic.
I think it'd be easier to bring up the task configs endpoint for MM2 than to
re-architect the Connect framework, especially given the compatibility and
migration concerns that would have to be addressed in order to allow non-leader
workers to write to the config topic. But either approach would work.
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17110047#comment-17110047
]
victor commented on KAFKA-9981:
---
[~ChrisEgerton] [~ryannedolan] hi.
In case of dedicated mm2 clusters, If the configBackingStore task is hosted on
a follower node，Can the following node write directly into the config topic?
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17108867#comment-17108867
]
Chris Egerton commented on KAFKA-9981:
--
[~ryannedolan] I think these configuration updates come from the connector
requesting task reconfiguration from the framework:
[https://github.com/apache/kafka/blob/62fa8fc9a95d738780d1f73d2d758d7329828feb/connect/mirror/src/main/java/org/apache/kafka/connect/mirror/MirrorSourceConnector.java#L232]
In distributed mode, this causes the framework to generate new task configs
from the connector and then, if they've changed, try to write them to the
config topic. However, only the leader is allowed to write directly to the
config topic, so if the connector is hosted on a follower node, then the node
has to forward those configs to the leader via the REST API:
[https://github.com/apache/kafka/blob/62fa8fc9a95d738780d1f73d2d758d7329828feb/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/distributed/DistributedHerder.java#L1316-L1340]
The endpoint for receiving these task configs was the subject of KIP-507, which
sought to close a security loophole that it presented at the time. You can see
the code for that internal endpoint here:
[https://github.com/apache/kafka/blob/62fa8fc9a95d738780d1f73d2d758d7329828feb/connect/runtime/src/main/java/org/apache/kafka/connect/runtime/rest/resources/ConnectorsResource.java#L268-L278]
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17108701#comment-17108701
]
Ryanne Dolan commented on KAFKA-9981:
-
Configuration updates only come from the REST API, afaik, which doesn't exist
when running with connect-mirror-maker.sh. So I'm not sure what would be
triggering the logic in the PR. In order to configuration changes to be picked
up at all, the leader must be restarted. Generally you can't know which nodes
are leaders of which flows, so generally it makes sense to just restart
everything with a new config.
This would change if we added back the REST API to connect-mirror-maker.sh (it
is purposefully turned off at present). If we had a REST API, _then_
configuration could change and workers would need to notify their leaders. But
that is not the case now.
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17107816#comment-17107816
]
victor commented on KAFKA-9981:
---
A new topic/partition is created, but the data is not synchronized.
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
[jira] [Commented] (KAFKA-9981) Running a dedicated mm2 cluster with more than one nodes,When the configuration is updated the task is not aware and will lose the update operation.
[
https://issues.apache.org/jira/browse/KAFKA-9981?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel=17106865#comment-17106865
]
Andre Price commented on KAFKA-9981:
Not sure if related but I think I'm seeing issues where task configs don't seem
to be getting updated consistently when new topics/partitions are found.
> Running a dedicated mm2 cluster with more than one nodes,When the
> configuration is updated the task is not aware and will lose the update
> operation.
>
>
> Key: KAFKA-9981
> URL: https://issues.apache.org/jira/browse/KAFKA-9981
> Project: Kafka
> Issue Type: Bug
> Components: mirrormaker
>Affects Versions: 2.4.0, 2.5.0, 2.4.1
>Reporter: victor
>Priority: Major
>
> DistributedHerder.reconfigureConnector induction config update as follows:
> {code:java}
> if (changed) {
> List> rawTaskProps = reverseTransform(connName,
> configState, taskProps);
> if (isLeader()) {
> configBackingStore.putTaskConfigs(connName, rawTaskProps);
> cb.onCompletion(null, null);
> } else {
> // We cannot forward the request on the same thread because this
> reconfiguration can happen as a result of connector
> // addition or removal. If we blocked waiting for the response from
> leader, we may be kicked out of the worker group.
> forwardRequestExecutor.submit(new Runnable() {
> @Override
> public void run() {
> try {
> String leaderUrl = leaderUrl();
> if (leaderUrl == null || leaderUrl.trim().isEmpty()) {
> cb.onCompletion(new ConnectException("Request to
> leader to " +
> "reconfigure connector tasks failed " +
> "because the URL of the leader's REST
> interface is empty!"), null);
> return;
> }
> String reconfigUrl = RestServer.urlJoin(leaderUrl,
> "/connectors/" + connName + "/tasks");
> log.trace("Forwarding task configurations for connector
> {} to leader", connName);
> RestClient.httpRequest(reconfigUrl, "POST", null,
> rawTaskProps, null, config, sessionKey, requestSignatureAlgorithm);
> cb.onCompletion(null, null);
> } catch (ConnectException e) {
> log.error("Request to leader to reconfigure connector
> tasks failed", e);
> cb.onCompletion(e, null);
> }
> }
> });
> }
> }
> {code}
> KafkaConfigBackingStore task checks for configuration updates,such as topic
> whitelist update.If KafkaConfigBackingStore task is not running on leader
> node,an HTTP request will be send to notify the leader of the configuration
> update.However,dedicated mm2 cluster does not have the HTTP server turned
> on,so the request will fail to be sent,causing the update operation to be
> lost.
--
This message was sent by Atlassian Jira
(v8.3.4#803005)
