Poll Interval for Kafka Connect Source

[Shiti Saxena]

Hi,

Is there a configuration to set poll Interval for a SourceTask using the
Connect API?
The JDBC Connector uses a custom property - poll.interval.ms but is there
an internal property which can be used by different connectors?

Thanks,
Shiti

Possible bug in Kafka Connect - Schema modified internally

[Shiti Saxena]

Hi,

When defining a Kafka pipeline I realized that the Schema of SinkRecord was
not the same as its SourceRecord. I used the following code snippet.

for(Field x :record.valueSchema().fields()){
outputStream.println(x.name());
}


Is this the correct way to get the schema or am I missing something?

SourceRecord schema was

id
name
email
department
modified
org.apache.kafka.connect.data.Struct@59d634b7


but it's corresponding SinkRecord Schema was

id
department
department
department
modified
org.apache.kafka.connect.data.Struct@...


Could someone please help me with getting the correct Schema? Or is this a
bug in the Connect API?

Kafka Connect - SinkRecord schema

[Shiti Saxena]

Hi,

I was trying to define the following Kafka Connect pipeline :
JDBC Source -> Console Sink  using bulk mode

I realized the schema resulting from SinkRecord.valueSchema was incorrect.
I modified FileStreamSinkTask's put method to,

public void put(Collection sinkRecords) {
for (SinkRecord record : sinkRecords) {
log.trace("Writing line to {}: {}", logFilename(), record.value());
for(Field x :record.valueSchema().fields()){
outputStream.println(x.name());
}
outputStream.println(record.value());
}
}

Is this the correct way to get the schema or am I missing something?

I was expecting the output to be

id
name
email
department
modified
org.apache.kafka.connect.data.Struct@59d634b7


but it was

id
department
department
department
modified
org.apache.kafka.connect.data.Struct@59d634b7


I tried it with different tables and realized that the names of
columns with same datatype are messed up.

Can someone please tell me how to get the schema?

Or is this is a bug in the Connect API since the schema of the
SourceRecord created by JDBC Source Connector is correct.

Re: Kafka Connect usage

[Shiti Saxena]

Hi Alex,

I am using the default files.

Do we need to start Kafka server and zookeeper separately before starting
Kafka connect?

Thanks,
Shiti
On 12 Jan 2016 23:11, "Alex Loddengaard" <a...@confluent.io> wrote:

> Hi Shiti, I'm not able to reproduce the problem with the default
> *.properties files you're passing to connect-standalone.sh. Can you share
> these three files?
>
> Thanks,
>
> Alex
>
> On Mon, Jan 11, 2016 at 10:14 PM, Shiti Saxena <ssaxena@gmail.com>
> wrote:
>
> > Hi,
> >
> > I tried executing the following,
> >
> > bin/connect-standalone.sh config/connect-standalone.properties
> > config/connect-file-source.properties
> > config/connect-console-sink.properties
> >
> > I created a file text.txt in kafka directory but get the error,
> >
> > ERROR Failed to flush WorkerSourceTask{id=local-file-source-0}, timed out
> > while waiting for producer to flush outstanding messages, 1 left
> > ({ProducerRecord(topic=connect-test, partition=null, key=[B@71bb594d,
> > value=[B@7f7ca90a=ProducerRecord(topic=connect-test, partition=null,
> > key=[B@71bb594d, value=[B@7f7ca90a})
> > (org.apache.kafka.connect.runtime.WorkerSourceTask:237)
> > [2016-01-12 11:43:51,948] ERROR Failed to commit offsets for
> > WorkerSourceTask{id=local-file-source-0}
> > (org.apache.kafka.connect.runtime.SourceTaskOffsetCommitter:112)
> >
> > Is any other configuration required?
> >
> > Thanks,
> > Shiti
> >
>
>
>
> --
> *Alex Loddengaard | **Solutions Architect | Confluent*
> *Download Apache Kafka and Confluent Platform: www.confluent.io/download
> <http://www.confluent.io/download>*
>

Kafka Connect Embedded API

[Shiti Saxena]

Hi,

I wanted to use Kafka Connect in an application and wanted to if the
Embedded API discussed in KIP-26

available
in 0.9.0.0 or is there an alternative?

Thanks,
Shiti

Programmable API for Kafka Connect ?

[Shiti Saxena]

Hi,

Does Kafka Connect have an API which can be used by applications to start
Kafka Connect, add/remove Connectors?

I also do not want to use the REST API and do not want to start the REST
server.

Thanks,
Shiti