[jira] [Created] (KAFKA-13485) Restart connectors after RetriableException raised from Task::start()
Gunnar Morling created KAFKA-13485:
--
Summary: Restart connectors after RetriableException raised from
Task::start()
Key: KAFKA-13485
URL: https://issues.apache.org/jira/browse/KAFKA-13485
Project: Kafka
Issue Type: Improvement
Components: KafkaConnect
Reporter: Gunnar Morling
If a {{RetriableException}} is raised from {{Task::start()}}, this doesn't
trigger an attempt to start that connector again. I.e. the restart
functionality currently is only implemented for exceptions raised from
{{poll()}}/{{put()}}. Triggering restarts also upon failures during {{start()}}
would be desirable, so to circumvent temporary failure conditions like a
network hickup which currrently require a manual restart of the affected tasks,
if a connector for instance establishes a database connection during
{{start()}}.
--
This message was sent by Atlassian Jira
(v8.20.1#820001)
[jira] [Created] (KAFKA-13478) KIP-802: Validation Support for Kafka Connect SMT Options
Gunnar Morling created KAFKA-13478: -- Summary: KIP-802: Validation Support for Kafka Connect SMT Options Key: KAFKA-13478 URL: https://issues.apache.org/jira/browse/KAFKA-13478 Project: Kafka Issue Type: Bug Components: KafkaConnect Reporter: Gunnar Morling Implement [KIP-802|KIP-802: Validation Support for Kafka Connect SMT Options], adding validation support for SMT options. -- This message was sent by Atlassian Jira (v8.20.1#820001)
[jira] [Created] (KAFKA-12806) KRaft: Confusing leadership status exposed for controller without quorum
Gunnar Morling created KAFKA-12806: -- Summary: KRaft: Confusing leadership status exposed for controller without quorum Key: KAFKA-12806 URL: https://issues.apache.org/jira/browse/KAFKA-12806 Project: Kafka Issue Type: Bug Affects Versions: 2.8.0 Reporter: Gunnar Morling -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-12801) High CPU load after restarting brokers subsequent to quorum loss
Gunnar Morling created KAFKA-12801: -- Summary: High CPU load after restarting brokers subsequent to quorum loss Key: KAFKA-12801 URL: https://issues.apache.org/jira/browse/KAFKA-12801 Project: Kafka Issue Type: Bug Components: core, KafkaConnect Affects Versions: 2.8.0 Reporter: Gunnar Morling I'm testing Kafka in the new KRaft mode added in 2.8. I have a cluster of three Kafka nodes (all combined nodes), and one Kafka Connect node. After starting all components, I first stop the current controller of the Kafka cluster, then I stop the then controller of the Kafka cluster. At this point, only one Kafka node out of the original three and Connect is running. When now restarting the two stopped Kafka nodes, CPU load on the Connect node and the two broker nodes goes up to 100% and remains at that level for an indefinite amount of time. -- This message was sent by Atlassian Jira (v8.3.4#803005)
[jira] [Created] (KAFKA-8523) InsertField transformation fails when encountering tombstone event
Gunnar Morling created KAFKA-8523:
-
Summary: InsertField transformation fails when encountering
tombstone event
Key: KAFKA-8523
URL: https://issues.apache.org/jira/browse/KAFKA-8523
Project: Kafka
Issue Type: Bug
Components: KafkaConnect
Reporter: Gunnar Morling
When applying the {{InsertField}} transformation to a tombstone event, an
exception is raised:
{code}
org.apache.kafka.connect.errors.DataException: Only Map objects supported in
absence of schema for [field insertion], found: null
at
org.apache.kafka.connect.transforms.util.Requirements.requireMap(Requirements.java:38)
at
org.apache.kafka.connect.transforms.InsertField.applySchemaless(InsertField.java:138)
at
org.apache.kafka.connect.transforms.InsertField.apply(InsertField.java:131)
at
org.apache.kafka.connect.transforms.InsertFieldTest.tombstone(InsertFieldTest.java:128)
{code}
AFAICS, the transform can still be made working in in this case by simply
building up a new value map from scratch.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-8476) Kafka 2.2.1 distribution contains JAX-RS API twice
Gunnar Morling created KAFKA-8476: - Summary: Kafka 2.2.1 distribution contains JAX-RS API twice Key: KAFKA-8476 URL: https://issues.apache.org/jira/browse/KAFKA-8476 Project: Kafka Issue Type: Bug Reporter: Gunnar Morling In kafka_2.12-2.2.1.tgz there is both javax.ws.rs-api-2.1.jar and javax.ws.rs-api-2.1.1.jar. I reckon only one should be there. -- This message was sent by Atlassian JIRA (v7.6.3#76005)
[jira] [Created] (KAFKA-7336) Kafka Connect source task when producing record with invalid topic name
Gunnar Morling created KAFKA-7336:
-
Summary: Kafka Connect source task when producing record with
invalid topic name
Key: KAFKA-7336
URL: https://issues.apache.org/jira/browse/KAFKA-7336
Project: Kafka
Issue Type: Bug
Components: KafkaConnect
Affects Versions: 2.0.0
Reporter: Gunnar Morling
If a Kafka Connect source task returns a {{SourceRecord}} with an invalid topic
name (e.g. "dbserver1.inventory.test@data"), that source task hangs (presumably
indefinitely?) and doesn't continue it's polling loop. The log is flooded with
this message:
{code}
connect_1| 2018-08-24 08:47:29,014 WARN || [Producer
clientId=producer-4] Error while fetching metadata with correlation id 833 :
{dbserver1.inventory.test@data=INVALID_TOPIC_EXCEPTION}
[org.apache.kafka.clients.NetworkClient]
{code}
The producer thread is stuck in the loop here:
{code}
KafkaProducer.waitOnMetadata(String, Integer, long) line: 938
KafkaProducer.doSend(ProducerRecord, Callback) line: 823
KafkaProducer.send(ProducerRecord, Callback) line: 803
WorkerSourceTask.sendRecords() line: 318
WorkerSourceTask.execute() line: 228
WorkerSourceTask(WorkerTask).doRun() line: 175
WorkerSourceTask(WorkerTask).run() line: 219
Executors$RunnableAdapter.call() line: 511
FutureTask.run() line: 266
ThreadPoolExecutor.runWorker(ThreadPoolExecutor$Worker) line: 1149
ThreadPoolExecutor$Worker.run() line: 624
Thread.run() line: 748
{code}
This causes the task to remain in RUNNING state, but no further invocations of
{{poll()}} are done.
Of course we'll work around this and make sure to not produce records with
invalid topic names, but I think the source task should transition to FAILED
state in this case.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-7058) ConnectSchema#equals() broken for array-typed default values
Gunnar Morling created KAFKA-7058:
-
Summary: ConnectSchema#equals() broken for array-typed default
values
Key: KAFKA-7058
URL: https://issues.apache.org/jira/browse/KAFKA-7058
Project: Kafka
Issue Type: Bug
Reporter: Gunnar Morling
{ConnectSchema#equals()} calls {{Objects#equals()}} for the schemas' default
values, but this doesn't work correctly if the default values in fact are
arrays. In this case, always {false} will be returned, also if the default
value arrays actually are the same.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-6566) SourceTask#stop() not called after exception raised in poll()
Gunnar Morling created KAFKA-6566:
-
Summary: SourceTask#stop() not called after exception raised in
poll()
Key: KAFKA-6566
URL: https://issues.apache.org/jira/browse/KAFKA-6566
Project: Kafka
Issue Type: Bug
Reporter: Gunnar Morling
Having discussed this with [~rhauch], it has been my assumption that
{{SourceTask#stop()}} will be called by the Kafka Connect framework in case an
exception has been raised in {{poll()}}. That's not the case, though. As an
example see the connector and task below.
Calling {{stop()}} after an exception in {{poll()}} seems like a very useful
action to take, as it'll allow the task to clean up any resources such as
releasing any database connections, right after that failure and not only once
the connector is stopped.
{code}
package com.example;
import java.util.Collections;
import java.util.List;
import java.util.Map;
import org.apache.kafka.common.config.ConfigDef;
import org.apache.kafka.connect.connector.Task;
import org.apache.kafka.connect.source.SourceConnector;
import org.apache.kafka.connect.source.SourceRecord;
import org.apache.kafka.connect.source.SourceTask;
public class TestConnector extends SourceConnector {
@Override
public String version() {
return null;
}
@Override
public void start(Map props) {
}
@Override
public Class taskClass() {
return TestTask.class;
}
@Override
public List> taskConfigs(int maxTasks) {
return Collections.singletonList(Collections.singletonMap("foo",
"bar"));
}
@Override
public void stop() {
}
@Override
public ConfigDef config() {
return new ConfigDef();
}
public static class TestTask extends SourceTask {
@Override
public String version() {
return null;
}
@Override
public void start(Map props) {
}
@Override
public List poll() throws InterruptedException {
throw new RuntimeException();
}
@Override
public void stop() {
System.out.println("stop() called");
}
}
}
{code}
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-6551) Unbounded queues in WorkerSourceTask cause OutOfMemoryError
Gunnar Morling created KAFKA-6551:
-
Summary: Unbounded queues in WorkerSourceTask cause
OutOfMemoryError
Key: KAFKA-6551
URL: https://issues.apache.org/jira/browse/KAFKA-6551
Project: Kafka
Issue Type: Bug
Components: KafkaConnect
Reporter: Gunnar Morling
A Debezium user reported an {{OutOfMemoryError}} to us, with over 50,000
messages in the {{WorkerSourceTask#outstandingMessages}} map.
This map is unbounded and I can't see any way of "rate limiting" which would
control how many records are added to it. Growth can only indirectly be limited
by reducing the offset flush interval, but as connectors can return large
amounts of messages in single {{poll()}} calls that's not sufficient in all
cases. Note the user reported this issue during snapshotting a database, i.e. a
high number of records arrived in a very short period of time.
To solve the problem I'd suggest to make this map backpressure-aware and thus
prevent its indefinite growth, so that no further records will be polled from
the connector until messages have been taken out of the map again.
--
This message was sent by Atlassian JIRA
(v7.6.3#76005)
[jira] [Created] (KAFKA-6456) Improve JavaDoc of SourceTask
Gunnar Morling created KAFKA-6456: - Summary: Improve JavaDoc of SourceTask Key: KAFKA-6456 URL: https://issues.apache.org/jira/browse/KAFKA-6456 Project: Kafka Issue Type: Improvement Components: KafkaConnect Affects Versions: 1.0.0 Reporter: Gunnar Morling -- This message was sent by Atlassian JIRA (v7.6.3#76005)
