Re: Link read avro from Kafka Connect Issue

[Tauzell, Dave]

Is Kafka connect adding some bytes to the beginning of the avro with the scheme 
registry id?

Dave

> On Nov 2, 2016, at 18:43, Will Du  wrote:
>
> By using the kafka-avro-console-consumer I am able to get rich message from 
> kafka connect with AvroConvert, but it got no output except schema from Flink
>
> By using the producer with defaultEncoding, the kafka-avro-console-consumer 
> throws exceptions show how. But Flink consumer works. But my target is to get 
> Flink costume avro data produced by Kafka connect
>
>> On Nov 2, 2016, at 7:36 PM, Will Du  wrote:
>>
>>
>> On Nov 2, 2016, at 7:31 PM, Will Du > > wrote:
>>
>> Hi folks,
>> I am trying to consume avro data from Kafka in Flink. The data is produced 
>> by Kafka connect using AvroConverter. I have created a 
>> AvroDeserializationSchema.java 
>>  used 
>> by Flink consumer. Then, I use following code to read it.
>>
>> public static void main(String[] args) throws Exception {
>>  StreamExecutionEnvironment env = 
>> StreamExecutionEnvironment.getExecutionEnvironment();
>>  Properties properties = new Properties();
>>  properties.setProperty("bootstrap.servers", “localhost:9092");
>>  properties.setProperty("zookeeper.connect", “localhost:2181”);
>> Schema schema = new Parser().parse("{" + "\"name\": \"test\", "
>>   + "\"type\": \"record\", "
>>   + "\"fields\": "
>>   +" [ "
>>   + "  { \"name\": \"name\", \"type\": 
>> \"string\" },"
>>   + "  { \"name\": \"symbol\", \"type\": 
>> \"string\" },"
>>   + "  { \"name\": \"exchange\", 
>> \"type\": \"string\"}"
>>   + "] "
>>   +"}");
>>
>>  AvroDeserializationSchema avroSchema = new 
>> AvroDeserializationSchema<>(schema);
>>  FlinkKafkaConsumer09 kafkaConsumer =
>>  new FlinkKafkaConsumer09<>("myavrotopic",avroSchema, 
>> properties);
>>  DataStream messageStream = 
>> env.addSource(kafkaConsumer);
>>  messageStream.rebalance().print();
>>  env.execute("Flink AVRO KAFKA Test");
>> }
>>
>> Once, I run the code, I am able to get the schema information only as 
>> follows.
>> {"name":"", "symbol":"", "exchange":""}
>> {"name":"", "symbol":"", "exchange":""}
>> {"name":"", "symbol":"", "exchange":""}
>> {"name":"", "symbol":"", "exchange":”"}
>>
>> Could anyone help to find out the issues why I cannot decode it?
>>
>> Further troubleshooting, I found out if I use a kafka producer here 
>>  to 
>> send the avro data especially using kafka.serializer.DefaultEncoder. Above 
>> code can get correct result. Does any body know how to either set 
>> DefaultEncoder in Kafka Connect or set it when writing customized kafka 
>> connect? Or in the other way, how should I modify the 
>> AvroDeserializationSchema.java for instead?
>>
>> Thanks, I’ll post this to the Flink user group as well.
>> Will
>
This e-mail and any files transmitted with it are confidential, may contain 
sensitive information, and are intended solely for the use of the individual or 
entity to whom they are addressed. If you have received this e-mail in error, 
please notify the sender by reply e-mail immediately and destroy all copies of 
the e-mail and any attachments.

Re: consumer_offsets partition skew and possibly ignored retention

[James Brown]

Jeff: This was with 0.9.0.1. It has not recurred since upgrading to
0.10.1.0.

On Fri, Oct 28, 2016 at 9:28 PM, Jeff Widman  wrote:

> James,
> What version did you experience the problem with?
>
> On Oct 28, 2016 6:26 PM, "James Brown"  wrote:
>
> > I was having this problem with one of my __consumer_offsets partitions; I
> > used reassignment to move the large partition onto a different set of
> > machines (which forced the cleaner to run through them again) and after
> the
> > new machines finished replicating, the partition was back down from 41GB
> to
> > a nice trim 38MB.
> >
> > On Fri, Oct 28, 2016 at 1:00 PM, Chi Hoang  wrote:
> >
> > > Hi,
> > > We have a 3-node cluster that is running 0.9.0.1, and recently saw that
> > the
> > > "__consumer_offsets" topic on one of the nodes seems really skewed with
> > > disk usage that looks like:
> > >
> > > 73G ./__consumer_offsets-10
> > > 0   ./__consumer_offsets-7
> > > 0   ./__consumer_offsets-4
> > > 0   ./__consumer_offsets-1
> > > 0   ./__consumer_offsets-49
> > > 19G ./__consumer_offsets-46
> > > 0   ./__consumer_offsets-43
> > > 0   ./__consumer_offsets-40
> > >
> > >
> > >
> > > This goes on for all 50 partitions.  Upon inspection, we saw that a lot
> > of
> > > the log files were old:
> > >
> > > ll __consumer_offsets-10
> > > total 76245192
> > > -rw-r--r-- 1 root root 0 Oct  7 20:14
> .index
> > > -rw-r--r-- 1 root root   901 Oct  7 20:14 .log
> > > -rw-r--r-- 1 root root157904 Oct  7 22:15
> 000907046457.index
> > > -rw-r--r-- 1 root root 104855056 Oct  7 22:15 000907046457.log
> > > -rw-r--r-- 1 root root157904 Oct  7 22:51
> 000909543421.index
> > > -rw-r--r-- 1 root root 104853568 Oct  7 22:51 000909543421.log
> > > -rw-r--r-- 1 root root157904 Oct  7 23:27
> 000910806717.index
> > > -rw-r--r-- 1 root root 104853568 Oct  7 23:27 000910806717.log
> > >
> > >
> > > We are using default parameters as it pertains to offset management,
> and
> > > our config output includes the following entries:
> > >
> > > log.cleaner.enable = true
> > >
> > > offsets.retention.minutes = 1440
> > >
> > >
> > > I tried looking through the issues on JIRA but didn't see a reported
> > > issue.  Does anyone know what's going on, and how I can fix this?
> > >
> > > Thanks.
> > >
> >
> >
> >
> > --
> > James Brown
> > Engineer
> >
>



-- 
James Brown
Engineer

Re: Link read avro from Kafka Connect Issue

[Will Du]

By using the kafka-avro-console-consumer I am able to get rich message from 
kafka connect with AvroConvert, but it got no output except schema from Flink 

By using the producer with defaultEncoding, the kafka-avro-console-consumer 
throws exceptions show how. But Flink consumer works. But my target is to get 
Flink costume avro data produced by Kafka connect

> On Nov 2, 2016, at 7:36 PM, Will Du  wrote:
> 
> 
> On Nov 2, 2016, at 7:31 PM, Will Du  > wrote:
> 
> Hi folks,
> I am trying to consume avro data from Kafka in Flink. The data is produced by 
> Kafka connect using AvroConverter. I have created a 
> AvroDeserializationSchema.java 
>  used by 
> Flink consumer. Then, I use following code to read it.
> 
> public static void main(String[] args) throws Exception {
>   StreamExecutionEnvironment env = 
> StreamExecutionEnvironment.getExecutionEnvironment();
>   Properties properties = new Properties();
>   properties.setProperty("bootstrap.servers", “localhost:9092");
>   properties.setProperty("zookeeper.connect", “localhost:2181”);
> Schema schema = new Parser().parse("{" + "\"name\": \"test\", "
>  + "\"type\": \"record\", "
>  + "\"fields\": "
>  +" [ "
>  + "  { \"name\": \"name\", \"type\": 
> \"string\" },"
>  + "  { \"name\": \"symbol\", 
> \"type\": \"string\" },"
>  + "  { \"name\": \"exchange\", 
> \"type\": \"string\"}"
>  + "] "
>  +"}");
> 
>   AvroDeserializationSchema avroSchema = new 
> AvroDeserializationSchema<>(schema);
>   FlinkKafkaConsumer09 kafkaConsumer = 
>   new FlinkKafkaConsumer09<>("myavrotopic",avroSchema, 
> properties);
>   DataStream messageStream = 
> env.addSource(kafkaConsumer);
>   messageStream.rebalance().print();
>   env.execute("Flink AVRO KAFKA Test");
> }
> 
> Once, I run the code, I am able to get the schema information only as follows.
> {"name":"", "symbol":"", "exchange":""}
> {"name":"", "symbol":"", "exchange":""}
> {"name":"", "symbol":"", "exchange":""}
> {"name":"", "symbol":"", "exchange":”"}
> 
> Could anyone help to find out the issues why I cannot decode it?
> 
> Further troubleshooting, I found out if I use a kafka producer here 
>  to send 
> the avro data especially using kafka.serializer.DefaultEncoder. Above code 
> can get correct result. Does any body know how to either set DefaultEncoder 
> in Kafka Connect or set it when writing customized kafka connect? Or in the 
> other way, how should I modify the AvroDeserializationSchema.java for instead?
> 
> Thanks, I’ll post this to the Flink user group as well.
> Will

Link read avro from Kafka Connect Issue

[Will Du]

On Nov 2, 2016, at 7:31 PM, Will Du  wrote:

Hi folks,
I am trying to consume avro data from Kafka in Flink. The data is produced by 
Kafka connect using AvroConverter. I have created a 
AvroDeserializationSchema.java 
 used by 
Flink consumer. Then, I use following code to read it.

public static void main(String[] args) throws Exception {
  StreamExecutionEnvironment env = 
StreamExecutionEnvironment.getExecutionEnvironment();
  Properties properties = new Properties();
  properties.setProperty("bootstrap.servers", “localhost:9092");
  properties.setProperty("zookeeper.connect", “localhost:2181”);
Schema schema = new Parser().parse("{" + "\"name\": \"test\", "
   + "\"type\": \"record\", "
   + "\"fields\": "
   +" [ "
   + "  { \"name\": \"name\", \"type\": 
\"string\" },"
   + "  { \"name\": \"symbol\", 
\"type\": \"string\" },"
   + "  { \"name\": \"exchange\", 
\"type\": \"string\"}"
   + "] "
   +"}");

  AvroDeserializationSchema avroSchema = new 
AvroDeserializationSchema<>(schema);
  FlinkKafkaConsumer09 kafkaConsumer = 
new FlinkKafkaConsumer09<>("myavrotopic",avroSchema, 
properties);
  DataStream messageStream = 
env.addSource(kafkaConsumer);
  messageStream.rebalance().print();
  env.execute("Flink AVRO KAFKA Test");
}

Once, I run the code, I am able to get the schema information only as follows.
{"name":"", "symbol":"", "exchange":""}
{"name":"", "symbol":"", "exchange":""}
{"name":"", "symbol":"", "exchange":""}
{"name":"", "symbol":"", "exchange":”"}

Could anyone help to find out the issues why I cannot decode it?

Further troubleshooting, I found out if I use a kafka producer here 
 to send 
the avro data especially using kafka.serializer.DefaultEncoder. Above code can 
get correct result. Does any body know how to either set DefaultEncoder in 
Kafka Connect or set it when writing customized kafka connect? Or in the other 
way, how should I modify the AvroDeserializationSchema.java for instead?

Thanks, I’ll post this to the Flink user group as well.
Will

Re: consumer_offsets partition skew and possibly ignored retention

[Chi Hoang]

I tried running reassignment on the topic, but that didn't help.  I had to
restart the broker for it to release the file handlers, then manually
delete.

On Fri, Oct 28, 2016 at 6:25 PM, James Brown  wrote:

> I was having this problem with one of my __consumer_offsets partitions; I
> used reassignment to move the large partition onto a different set of
> machines (which forced the cleaner to run through them again) and after the
> new machines finished replicating, the partition was back down from 41GB to
> a nice trim 38MB.
>
> On Fri, Oct 28, 2016 at 1:00 PM, Chi Hoang  wrote:
>
> > Hi,
> > We have a 3-node cluster that is running 0.9.0.1, and recently saw that
> the
> > "__consumer_offsets" topic on one of the nodes seems really skewed with
> > disk usage that looks like:
> >
> > 73G ./__consumer_offsets-10
> > 0   ./__consumer_offsets-7
> > 0   ./__consumer_offsets-4
> > 0   ./__consumer_offsets-1
> > 0   ./__consumer_offsets-49
> > 19G ./__consumer_offsets-46
> > 0   ./__consumer_offsets-43
> > 0   ./__consumer_offsets-40
> >
> >
> >
> > This goes on for all 50 partitions.  Upon inspection, we saw that a lot
> of
> > the log files were old:
> >
> > ll __consumer_offsets-10
> > total 76245192
> > -rw-r--r-- 1 root root 0 Oct  7 20:14 .index
> > -rw-r--r-- 1 root root   901 Oct  7 20:14 .log
> > -rw-r--r-- 1 root root157904 Oct  7 22:15 000907046457.index
> > -rw-r--r-- 1 root root 104855056 Oct  7 22:15 000907046457.log
> > -rw-r--r-- 1 root root157904 Oct  7 22:51 000909543421.index
> > -rw-r--r-- 1 root root 104853568 Oct  7 22:51 000909543421.log
> > -rw-r--r-- 1 root root157904 Oct  7 23:27 000910806717.index
> > -rw-r--r-- 1 root root 104853568 Oct  7 23:27 000910806717.log
> >
> >
> > We are using default parameters as it pertains to offset management, and
> > our config output includes the following entries:
> >
> > log.cleaner.enable = true
> >
> > offsets.retention.minutes = 1440
> >
> >
> > I tried looking through the issues on JIRA but didn't see a reported
> > issue.  Does anyone know what's going on, and how I can fix this?
> >
> > Thanks.
> >
>
>
>
> --
> James Brown
> Engineer
>

Re: Kafka Streams fails permanently when used with an unstable network

[Eno Thereska]

Hi Sai,

For your second note on rebalancing taking a long time, we have just improved 
the situation in trunk after fixing this JIRA: 
https://issues.apache.org/jira/browse/KAFKA-3559 
. Feel free to give it a go 
if rebalancing time continues to be a problem.

Thanks
Eno

> On 31 Oct 2016, at 19:44, saiprasad mishra  wrote:
> 
> Hey Guys
> 
> I have noticed similar issues when network goes down on starting of kafka
> stream apps especially the store has initialized but the task
> initialization is not complete and when the network comes back the
> rebalance fails with the above error and I had to restart. as i run many
> partitions and have many tasks get initialized.
> 
> Otherwise if the kafka streams app is started successfully does recover
> from network issues always as far as what I have seen so far and also
> stores do remain available.
> 
> Which means some of these initialization exceptions can be categorized as
> recoverable and should be always retried.
> 
> I think task 0_0 in your case was not initialized properly in the first
> place and then rebalance happened bcoz of network connectivity and it
> resulted in the above exception.
> 
> On a separate note rebalance takes longer time  as i have some
> intermeidiary topics and thinking it might be worse if network is slow and
> was thinking of something like store may be available for querying quickly
> without waiting for the full initialization of tasks
> 
> Regards
> Sai
> 
> 
> 
> 
> 
> 
> Regards
> Sai
> 
> On Mon, Oct 31, 2016 at 3:51 AM, Damian Guy  wrote:
> 
>> Hi Frank,
>> 
>> This usually means that another StreamThread has the lock for the state
>> directory. So it would seem that one of the StreamThreads hasn't shut down
>> cleanly. If it happens again can you please take a Thread Dump so we can
>> see what is happening?
>> 
>> Thanks,
>> Damian
>> 
>> On Sun, 30 Oct 2016 at 10:52 Frank Lyaruu  wrote:
>> 
>>> I have a remote Kafka cluster, to which I connect using a VPN and a
>>> not-so-great WiFi network.
>>> That means that sometimes the Kafka Client loses briefly loses
>>> connectivity.
>>> When it regains a connection after a while, I see:
>>> 
>>> org.apache.kafka.clients.consumer.CommitFailedException: Commit cannot
>> be
>>> completed since the group has already rebalanced and assigned the
>>> partitions to another member. This means that the time between subsequent
>>> calls to poll() was longer than the configured max.poll.interval.ms,
>> which
>>> typically implies that the poll loop is spending too much time message
>>> processing. You can address this either by increasing the session timeout
>>> or by reducing the maximum size of batches returned in poll() with
>>> max.poll.records.
>>> 
>>> ...
>>> 
>>> Which makes sense I suppose, but this shouldn't be fatal.
>>> 
>>> But then I see:
>>> [StreamThread-1] ERROR
>>> org.apache.kafka.streams.processor.internals.StreamThread -
>> stream-thread
>>> [StreamThread-1] Failed to create an active task %s:
>>> org.apache.kafka.streams.errors.ProcessorStateException: task [0_0]
>> Error
>>> while creating the state manager
>>> 
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.AbstractTask.(
>> AbstractTask.java:72)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.
>> StreamTask.(StreamTask.java:89)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.
>> StreamThread.createStreamTask(StreamThread.java:633)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.
>> StreamThread.addStreamTasks(StreamThread.java:660)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.StreamThread.access$100(
>> StreamThread.java:69)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.StreamThread$1.
>> onPartitionsAssigned(StreamThread.java:124)
>>> at
>>> 
>>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.
>> onJoinComplete(ConsumerCoordinator.java:228)
>>> at
>>> 
>>> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.
>> joinGroupIfNeeded(AbstractCoordinator.java:313)
>>> at
>>> 
>>> org.apache.kafka.clients.consumer.internals.AbstractCoordinator.
>> ensureActiveGroup(AbstractCoordinator.java:277)
>>> at
>>> 
>>> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.poll(
>> ConsumerCoordinator.java:259)
>>> at
>>> 
>>> org.apache.kafka.clients.consumer.KafkaConsumer.
>> pollOnce(KafkaConsumer.java:1013)
>>> at
>>> 
>>> org.apache.kafka.clients.consumer.KafkaConsumer.poll(
>> KafkaConsumer.java:979)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(
>> StreamThread.java:407)
>>> at
>>> 
>>> org.apache.kafka.streams.processor.internals.
>> StreamThread.run(StreamThread.java:242)
>>> 
>>> Caused by: java.io.IOException: task [0_0] Failed to lock the state
>>> directory:
>>> 
>>> /Users/frank/git/dexels.repository/com.dexels.kafka.
>> 

Re: Modify in-flight messages

[Eno Thereska]

Hi Dominik,

Not sure if this is 100% relevant, but since I noticed you saying that you are 
benchmarking stream processing engines, one way to modify a message would be to 
use the Kafka Streams library, where you consume a message from a topic, modify 
it as needed/do some processing, and then produce further to another topic.

Thanks
Eno

> On 1 Nov 2016, at 15:46, Dominik Safaric  wrote:
> 
> Dear Michael,
> 
> Thanks for sharing this information with me. 
> 
> I am aware of the fact that each message has a timestamp indicating either 
> the log append or creation time. 
> 
> But in my case, this is not enough since I want to derive the consumption 
> time of messages. The reason for this is because we are currently in the 
> process of an empirical research benchmarking several stream processing 
> engines. 
> 
> Anyhow, I’ll take a look at the intercept classes.
> 
> Dominik
> 
>> On 1 Nov 2016, at 14:40, Mickael Maison  wrote:
>> 
>> Hi Dominik,
>> 
>> On both the consumer and producer you can use the
>> "interceptor.classes" config to specify classes that intercept and can
>> modify records when they are sent/received.
>> Also as of Kafka 0.10, messages have a timestamp field. See
>> http://kafka.apache.org/0101/javadoc/org/apache/kafka/clients/consumer/ConsumerRecord.html#timestamp().
>> 
>> On Tue, Nov 1, 2016 at 12:17 PM, Dominik Safaric
>>  wrote:
>>> Is it possible to somehow modify the Kafka message payload before being 
>>> sent to the consumer for consumption? Such as for example adding a 
>>> timestamp to the current message payload indicating the time of message 
>>> consumption.
>>> 
>>> Dominik Šafarić
> 

Re : windowing with the processor api

[Hamza HACHANI]

Thanks a lot.
This was very helpful .

Hamza



- Message de réponse -
De : "Eno Thereska" 
Pour : "users@kafka.apache.org" 
Objet : windowing with the processor api
Date : mer., nov. 2, 2016 19:18

Thanks Matthias, yes, to get window operations, or things like hopping or 
sliding windows you need to use the DSL (e.g., TimeWindows class). The 
Processor API is very basic (and thus flexible but) you'll end up 
re-implementing TimeWindows.

Eno

> On 2 Nov 2016, at 17:45, Matthias J. Sax  wrote:
>
> -BEGIN PGP SIGNED MESSAGE-
> Hash: SHA512
>
> A windowed store does not work the way you expect it. The parameter
> "windowSize" is not a store parameter itself, but a caching parameter
> for the store (only used if caching get's enabled).
>
> For window support, Streams provide window semantics on top of the
> store and the store is not aware of window in this sense. Each window
> gets an ID that is encoded in the store key as "record-key:window-ID".
> And records timestamps are mapped to window-IDs to find the correct
> window a records gets put into... To the store is still a plain
> key-value store and is not aware of any windowing stuff.
>
> I would highly recommend to use DSL to use windows operations. This
> should not be a limitation as you can mix-and-match DSL and Processor
> API. All you can do with plain processor API you can also do within
> DSL via .process(...)
>
>
> - -Matthias
>
> On 11/2/16 3:49 AM, Hamza HACHANI wrote:
>> Hi Eno,
>>
>>
>> What I want to say is that i don't find a place where to define the
>> size of the window and where to precise the time of the advance.
>>
>>
>> Hamza
>>
>> Thanks
>>
>>  De : Eno Thereska
>>  Envoyé : mardi 1 novembre 2016 22:44:47 À
>> : users@kafka.apache.org Objet : Re: windowing with the processor
>> api
>>
>> Hi Hamza,
>>
>> Are you getting a particular error? Here is an example :
>>
>> Stores.create("window-store") .withStringKeys()
>> .withStringValues() .persistent() .windowed(10, 10, 2,
>> false).build(), "the-processor")
>>
>> Thanks Eno
>>
>>> On 2 Nov 2016, at 08:19, Hamza HACHANI 
>>> wrote:
>>>
>>> Hi,
>>>
>>> I would like to know if somebody has an idea how to define the
>>> size of the window in the processor api.
>>>
>>> I've been blocked for 6 days looking for a solution.
>>>
>>> using :
>>>
>>> Stores.create(...).withStringKeys().withStringValues().persistent().w
> indowed(...).build()
>>>
>>>
>>>
> I was able to define the retention time but not the the size of the wind
> ow.
>>>
>>> Please help me if possible.
>>>
>>> Thanks,
>>>
>>> Hamza
>>
>>
> -BEGIN PGP SIGNATURE-
> Comment: GPGTools - https://gpgtools.org
>
> iQIcBAEBCgAGBQJYGiZDAAoJECnhiMLycopPIJsQAJUmjdbgYctWfKQEmPGcgXip
> 0S7Aicm6g1A9imvb0i33J+wGBf8nx4oHHZz53UmXM/DxTf0JlOSqcxGMAM5YTw04
> IVs5r90E+MiuvaqVoWC0FHMOzZwiXx88Rr192jjrgg3KaoAvjo1WeuF0voD7iQK5
> 7eBcVh0jZELxKFMR51Ax9BQ24DQSbrnjX45Opcn8BZEbwDEp+FIZxQfZLRvu2HRK
> JFgu0ur8pNNvSw8QDJ2ivBXNZ9sEu1altlmpHlrpYw8N8KmJ2bQXQWZYNZZjDrrm
> OjUznGjFsaLGJJymKZiAji49x3anM+h35dzFbJyy0AhXG/mtU3wI5zyxQk8dyue2
> 3iHnehHfdIVj7/STbBSj9ZhcWpvFVotfpLz1Nst5lnN6geGuhBC3ZlHf14yInu/e
> 64rzgJHWrnaSOlicPdQ4b2Y+EH3rFfH2iVJfSG0fOL+QzcqFLcuZfnLJV4PJhqiR
> qB8mY4p0yTdNZIRgTDc3lyTuVv3+lxj810XCp7evhA3erkGDV/hc0lV8Fmqb5eFC
> O3Z3k1N7rlOT7R1ATioONr5JMFgATh0nkpglY91dG38F297PkUZzFpdVe/79gh84
> Z6CE1M32vompN65QQ5P4jB8V24Z0RmaLFhnAknZjZUHCwuLyFaYvJs3RAJVWRIBz
> AxqqweAPlcocjRt3DHu3
> =ZLY9
> -END PGP SIGNATURE-

Kafka Streams Error

[Furkan KAMACI]

I use Kafka 0.10.0.1. I count the messages of a topic as follows:

...
streamsConfiguration.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
"earliest");
...
KStream longs = builder.stream(Serdes.String(),
Serdes.String(), "qps-input");
...
KTable longCounts =
longs.countByKey(TimeWindows.of("qps", 3600 * 1000),
Serdes.String());
...

and then I write output to another topic. Result is that:

Numbers which starts from 1 and increase whenever I add something to
qps-input.

My questions:

1) Does it calculate really last hour or everything from the beginning due
you I've set it as earliest?

2) Sometimes it's been reset and numbers starts from 1. What can be the
reason for that?

Kind Regards,
Furkan KAMACI

Added to Wiki please

[Kenny Gorman]

Per the wiki, I am emailing the list for this. Can you please add us to 
https://cwiki.apache.org/confluence/display/KAFKA/Powered+By?

Eventador.io (https://www.eventador.io/) is a whole stack Kafka as-a-service 
company. We enable developers to quickly create and painlessly manage real-time 
data pipelines on Apache Kafka.

Thx!!
Kenny Gorman
Founder 
www.eventador.io

HDFS Connector Compression?

[Henry Kim]

Is it possible to add compression to the HDFS Connector out of the box? Or does 
it require code change?


Thanks

Henry Kim

Re: windowing with the processor api

[Eno Thereska]

Thanks Matthias, yes, to get window operations, or things like hopping or 
sliding windows you need to use the DSL (e.g., TimeWindows class). The 
Processor API is very basic (and thus flexible but) you'll end up 
re-implementing TimeWindows.

Eno

> On 2 Nov 2016, at 17:45, Matthias J. Sax  wrote:
> 
> -BEGIN PGP SIGNED MESSAGE-
> Hash: SHA512
> 
> A windowed store does not work the way you expect it. The parameter
> "windowSize" is not a store parameter itself, but a caching parameter
> for the store (only used if caching get's enabled).
> 
> For window support, Streams provide window semantics on top of the
> store and the store is not aware of window in this sense. Each window
> gets an ID that is encoded in the store key as "record-key:window-ID".
> And records timestamps are mapped to window-IDs to find the correct
> window a records gets put into... To the store is still a plain
> key-value store and is not aware of any windowing stuff.
> 
> I would highly recommend to use DSL to use windows operations. This
> should not be a limitation as you can mix-and-match DSL and Processor
> API. All you can do with plain processor API you can also do within
> DSL via .process(...)
> 
> 
> - -Matthias
> 
> On 11/2/16 3:49 AM, Hamza HACHANI wrote:
>> Hi Eno,
>> 
>> 
>> What I want to say is that i don't find a place where to define the
>> size of the window and where to precise the time of the advance.
>> 
>> 
>> Hamza
>> 
>> Thanks
>> 
>>  De : Eno Thereska
>>  Envoyé : mardi 1 novembre 2016 22:44:47 À
>> : users@kafka.apache.org Objet : Re: windowing with the processor
>> api
>> 
>> Hi Hamza,
>> 
>> Are you getting a particular error? Here is an example :
>> 
>> Stores.create("window-store") .withStringKeys() 
>> .withStringValues() .persistent() .windowed(10, 10, 2,
>> false).build(), "the-processor")
>> 
>> Thanks Eno
>> 
>>> On 2 Nov 2016, at 08:19, Hamza HACHANI 
>>> wrote:
>>> 
>>> Hi,
>>> 
>>> I would like to know if somebody has an idea how to define the
>>> size of the window in the processor api.
>>> 
>>> I've been blocked for 6 days looking for a solution.
>>> 
>>> using :
>>> 
>>> Stores.create(...).withStringKeys().withStringValues().persistent().w
> indowed(...).build()
>>> 
>>> 
>>> 
> I was able to define the retention time but not the the size of the wind
> ow.
>>> 
>>> Please help me if possible.
>>> 
>>> Thanks,
>>> 
>>> Hamza
>> 
>> 
> -BEGIN PGP SIGNATURE-
> Comment: GPGTools - https://gpgtools.org
> 
> iQIcBAEBCgAGBQJYGiZDAAoJECnhiMLycopPIJsQAJUmjdbgYctWfKQEmPGcgXip
> 0S7Aicm6g1A9imvb0i33J+wGBf8nx4oHHZz53UmXM/DxTf0JlOSqcxGMAM5YTw04
> IVs5r90E+MiuvaqVoWC0FHMOzZwiXx88Rr192jjrgg3KaoAvjo1WeuF0voD7iQK5
> 7eBcVh0jZELxKFMR51Ax9BQ24DQSbrnjX45Opcn8BZEbwDEp+FIZxQfZLRvu2HRK
> JFgu0ur8pNNvSw8QDJ2ivBXNZ9sEu1altlmpHlrpYw8N8KmJ2bQXQWZYNZZjDrrm
> OjUznGjFsaLGJJymKZiAji49x3anM+h35dzFbJyy0AhXG/mtU3wI5zyxQk8dyue2
> 3iHnehHfdIVj7/STbBSj9ZhcWpvFVotfpLz1Nst5lnN6geGuhBC3ZlHf14yInu/e
> 64rzgJHWrnaSOlicPdQ4b2Y+EH3rFfH2iVJfSG0fOL+QzcqFLcuZfnLJV4PJhqiR
> qB8mY4p0yTdNZIRgTDc3lyTuVv3+lxj810XCp7evhA3erkGDV/hc0lV8Fmqb5eFC
> O3Z3k1N7rlOT7R1ATioONr5JMFgATh0nkpglY91dG38F297PkUZzFpdVe/79gh84
> Z6CE1M32vompN65QQ5P4jB8V24Z0RmaLFhnAknZjZUHCwuLyFaYvJs3RAJVWRIBz
> AxqqweAPlcocjRt3DHu3
> =ZLY9
> -END PGP SIGNATURE-

Re: windowing with the processor api

[Matthias J. Sax]

-BEGIN PGP SIGNED MESSAGE-
Hash: SHA512

A windowed store does not work the way you expect it. The parameter
"windowSize" is not a store parameter itself, but a caching parameter
for the store (only used if caching get's enabled).

For window support, Streams provide window semantics on top of the
store and the store is not aware of window in this sense. Each window
gets an ID that is encoded in the store key as "record-key:window-ID".
And records timestamps are mapped to window-IDs to find the correct
window a records gets put into... To the store is still a plain
key-value store and is not aware of any windowing stuff.

I would highly recommend to use DSL to use windows operations. This
should not be a limitation as you can mix-and-match DSL and Processor
API. All you can do with plain processor API you can also do within
DSL via .process(...)


- -Matthias

On 11/2/16 3:49 AM, Hamza HACHANI wrote:
> Hi Eno,
> 
> 
> What I want to say is that i don't find a place where to define the
> size of the window and where to precise the time of the advance.
> 
> 
> Hamza
> 
> Thanks
> 
>  De : Eno Thereska
>  Envoyé : mardi 1 novembre 2016 22:44:47 À
> : users@kafka.apache.org Objet : Re: windowing with the processor
> api
> 
> Hi Hamza,
> 
> Are you getting a particular error? Here is an example :
> 
> Stores.create("window-store") .withStringKeys() 
> .withStringValues() .persistent() .windowed(10, 10, 2,
> false).build(), "the-processor")
> 
> Thanks Eno
> 
>> On 2 Nov 2016, at 08:19, Hamza HACHANI 
>> wrote:
>> 
>> Hi,
>> 
>> I would like to know if somebody has an idea how to define the
>> size of the window in the processor api.
>> 
>> I've been blocked for 6 days looking for a solution.
>> 
>> using :
>> 
>> Stores.create(...).withStringKeys().withStringValues().persistent().w
indowed(...).build()
>>
>>
>> 
I was able to define the retention time but not the the size of the wind
ow.
>> 
>> Please help me if possible.
>> 
>> Thanks,
>> 
>> Hamza
> 
> 
-BEGIN PGP SIGNATURE-
Comment: GPGTools - https://gpgtools.org

iQIcBAEBCgAGBQJYGiZDAAoJECnhiMLycopPIJsQAJUmjdbgYctWfKQEmPGcgXip
0S7Aicm6g1A9imvb0i33J+wGBf8nx4oHHZz53UmXM/DxTf0JlOSqcxGMAM5YTw04
IVs5r90E+MiuvaqVoWC0FHMOzZwiXx88Rr192jjrgg3KaoAvjo1WeuF0voD7iQK5
7eBcVh0jZELxKFMR51Ax9BQ24DQSbrnjX45Opcn8BZEbwDEp+FIZxQfZLRvu2HRK
JFgu0ur8pNNvSw8QDJ2ivBXNZ9sEu1altlmpHlrpYw8N8KmJ2bQXQWZYNZZjDrrm
OjUznGjFsaLGJJymKZiAji49x3anM+h35dzFbJyy0AhXG/mtU3wI5zyxQk8dyue2
3iHnehHfdIVj7/STbBSj9ZhcWpvFVotfpLz1Nst5lnN6geGuhBC3ZlHf14yInu/e
64rzgJHWrnaSOlicPdQ4b2Y+EH3rFfH2iVJfSG0fOL+QzcqFLcuZfnLJV4PJhqiR
qB8mY4p0yTdNZIRgTDc3lyTuVv3+lxj810XCp7evhA3erkGDV/hc0lV8Fmqb5eFC
O3Z3k1N7rlOT7R1ATioONr5JMFgATh0nkpglY91dG38F297PkUZzFpdVe/79gh84
Z6CE1M32vompN65QQ5P4jB8V24Z0RmaLFhnAknZjZUHCwuLyFaYvJs3RAJVWRIBz
AxqqweAPlcocjRt3DHu3
=ZLY9
-END PGP SIGNATURE-

Re: Question regarding dynamic subscriber environment

[Gerrit Jansen van Vuuren]

yes. You open a connection, and the fetch threads will look at a shared
variable for the topics to fetch, its this shared variable that is updated
when you add and remove topics. The connection itself is not closed. There
is no relation between a connection and the topics being consumed, other
than the connection is used to fetch data for a particular topic, for each
broker used a pool of connections is kept, and when a topic is fetched at
any point a connection for the required broker is selected from the pool,
then returned to the pool after use.

See: https://github.com/gerritjvv/kafka-fast#java-1

Consumer consumer = Consumer.connect(new KafkaConf(), new
BrokerConf[]{new BrokerConf("192.168.4.40", 9092)}, new
RedisConf("192.168.4.10", 6379, "test-group"), "my-topic");Message msg
= consumer.readMsg();

//Add topics
consumer.addTopics("topic1", "topic2");
//Remove topics
consumer.removeTopics("topic1", "topic2");



On Wed, Nov 2, 2016 at 4:37 PM, Janagan Sivagnanasundaram <
janagan1...@gmail.com> wrote:

> Does this really address the respective problem? Ultimate task is that, the
> connection between broker and subscriber should not be terminated.
> Subscriber is free to change his topic interests without closing the
> connection.
>
> On Wed, Nov 2, 2016 at 12:43 PM, Gerrit Jansen van Vuuren <
> gerrit...@gmail.com> wrote:
>
> > Hi,
> >
> > Have a look at the kafka client lib
> > https://github.com/gerritjvv/kafka-fast#java-1, it already provides this
> > functionality.
> >
> >
> > On Wed, Nov 2, 2016 at 2:34 AM, Janagan Sivagnanasundaram <
> > janagan1...@gmail.com> wrote:
> >
> > > Kafka's current nature is does not support to dynamic subscriber
> > > environment where the topic interest of the subscriber is keep change
> > > overtime unless subscriber terminate the particular connection and
> > > reconnect.
> > >
> > > What I have planned to implement or design is to enable the dynamic
> > > subscriber environment where the subscriber can change his topic
> > interests
> > > without terminating the connection with it's broker. Following are some
> > > ideas that we can use to do such thing.
> > >
> > > 1) Modifying the current protocol by adding an extra attribute
> > > 2) Create a control connection channel from subscriber to broker
> > >
> > > Is there easy/efficient way to do such thing?
> > >
> >
>

Re: Question regarding dynamic subscriber environment

[Janagan Sivagnanasundaram]

Does this really address the respective problem? Ultimate task is that, the
connection between broker and subscriber should not be terminated.
Subscriber is free to change his topic interests without closing the
connection.

On Wed, Nov 2, 2016 at 12:43 PM, Gerrit Jansen van Vuuren <
gerrit...@gmail.com> wrote:

> Hi,
>
> Have a look at the kafka client lib
> https://github.com/gerritjvv/kafka-fast#java-1, it already provides this
> functionality.
>
>
> On Wed, Nov 2, 2016 at 2:34 AM, Janagan Sivagnanasundaram <
> janagan1...@gmail.com> wrote:
>
> > Kafka's current nature is does not support to dynamic subscriber
> > environment where the topic interest of the subscriber is keep change
> > overtime unless subscriber terminate the particular connection and
> > reconnect.
> >
> > What I have planned to implement or design is to enable the dynamic
> > subscriber environment where the subscriber can change his topic
> interests
> > without terminating the connection with it's broker. Following are some
> > ideas that we can use to do such thing.
> >
> > 1) Modifying the current protocol by adding an extra attribute
> > 2) Create a control connection channel from subscriber to broker
> >
> > Is there easy/efficient way to do such thing?
> >
>

Re: connection closed by kafka

[Rajini Sivaram]

Broker closes client connections that are idle for a configurable period of
time (broker property connections.max.idle.ms). The default idle time is 10
minutes which matches the close time in the logs.

On Wed, Nov 2, 2016 at 2:43 PM, Jaikiran Pai 
wrote:

> Which exact version of Kafka installation and Kafka client is this? And
> which language/library of Kafka client? Also, are you describing this
> situation in the context of producing messages? Can you post your relevant
> code from the application where you deal with this?
>
> Connection management is an internal detail of Kafka client libraries and
> usually won't end up outside of it, so you shouldn't really notice any of
> these issues.
>
>
> -Jaikiran
>
>
> On Friday 28 October 2016 05:50 AM, Jianbin Wei wrote:
>
>> In our environment we notice that sometimes Kafka would close the
>> connection after one message is sent over.  The client does not detect that
>> and tries to send another message again.  That triggers a RST packet.
>>
>> Any idea why the Kafka broker would close the connection?
>>
>> Attached you can find the packets between our client and kafka broker.
>>
>>
>> 20:55:40.834543 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [S],
>> seq 31787730, win 14600, options [mss 1460,nop,nop,sackOK,nop,wscale 9],
>> length 0
>> 0x:  4500 0034 8cc1 4000 4006 ca67 ac12 45c2  E..4..@
>> .@..g..E.
>> 0x0010:  ac12 45b4 868d 2384 01e5 0ad2    ..E...#.
>> 0x0020:  8002 3908 e3c1  0204 05b4 0101 0402  ..9.
>> 0x0030:  0103 0309
>> 20:55:40.834744 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [S.],
>> seq 1238329644, ack 31787731, win 14600, options [mss
>> 1460,nop,nop,sackOK,nop,wscale 1], length 0
>> 0x:  4500 0034  4000 4006 5729 ac12 45b4  E..4..@
>> .@.W)..E.
>> 0x0010:  ac12 45c2 2384 868d 49cf 692c 01e5 0ad3  ..E.#...I.i,
>> 0x0020:  8012 3908 e89e  0204 05b4 0101 0402  ..9.
>> 0x0030:  0103 0301
>> 20:55:40.834787 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [.],
>> ack 1, win 29, length 0
>> 0x:  4500 0028 8cc2 4000 4006 ca72 ac12 45c2  E..(..@.@..r..E.
>> 0x0010:  ac12 45b4 868d 2384 01e5 0ad3 49cf 692d  ..E...#.I.i-
>> 0x0020:  5010 001d e3b5   P...
>> 20:55:40.834921 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [P.],
>> seq 1:691, ack 1, win 29, length 690
>> 0x:  4500 02da 8cc3 4000 4006 c7bf ac12 45c2  E.@
>> .@.E.
>> 0x0010:  ac12 45b4 868d 2384 01e5 0ad3 49cf 692d  ..E...#.I.i-
>> 0x0020:  5018 001d e667   02ae    Pg..
>> 0x0030:   0003 000c 6b61 666b 612d 7079 7468  ..kafka-pyth
>> 0x0040:  6f6e 0001  03e8  0001 000e 6576  onev
>> 0x0050:  656e 745f 6e73 706f 6c69 6379  0001  ent_nspolicy
>> 0x0060:     0272      ...r
>> 0x0070:   0266 4ff3 bd11   0004 3131  ...fO.11
>> 0x0080:  3238  0254 5b30 2c7b 2261 7022 3a22  28...T[0,{"ap":"
>>
>> 20:55:40.835297 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [.],
>> ack 691, win 7990, length 0
>> 0x:  4500 0028 e872 4000 4006 6ec2 ac12 45b4  E..(.r@
>> .@.n...E.
>> 0x0010:  ac12 45c2 2384 868d 49cf 692d 01e5 0d85  ..E.#...I.i-
>> 0x0020:  5010 1f36 408b       P..6@.
>> 20:55:40.837837 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [P.],
>> seq 1:47, ack 691, win 7990, length 46
>> 0x:  4500 0056 e873 4000 4006 6e93 ac12 45b4  E..V.s@
>> .@.n...E.
>> 0x0010:  ac12 45c2 2384 868d 49cf 692d 01e5 0d85  ..E.#...I.i-
>> 0x0020:  5018 1f36 ece3   002a  0003  P..6...*
>> 0x0030:   0001 000e 6576 656e 745f 6e73 706f  ..event_nspo
>> 0x0040:  6c69 6379  0001      licy
>> 0x0050:   0003 6527   e'
>> 20:55:40.837853 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [.],
>> ack 47, win 29, length 0
>> 0x:  4500 0028 8cc4 4000 4006 ca70 ac12 45c2  E..(..@.@..p..E.
>> 0x0010:  ac12 45b4 868d 2384 01e5 0d85 49cf 695b  ..E...#.I.i[
>> 0x0020:  5010 001d e3b5   P...
>>
>> Closed here
>
 21:05:40.839440 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags
>> [F.], seq 47, ack 691, win 7990, length 0
>> 0x:  4500 0028 e874 4000 4006 6ec0 ac12 45b4  E..(.t@
>> .@.n...E.
>> 0x0010:  ac12 45c2 2384 868d 49cf 695b 01e5 0d85  ..E.#...I.i[
>> 0x0020:  5011 1f36 405c       P..6@\
>> 21:05:40.876047 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [.],
>> ack 48, win 

Re: connection closed by kafka

[Jaikiran Pai]

Which exact version of Kafka installation and Kafka client is this? And 
which language/library of Kafka client? Also, are you describing this 
situation in the context of producing messages? Can you post your 
relevant code from the application where you deal with this?


Connection management is an internal detail of Kafka client libraries 
and usually won't end up outside of it, so you shouldn't really notice 
any of these issues.



-Jaikiran

On Friday 28 October 2016 05:50 AM, Jianbin Wei wrote:

In our environment we notice that sometimes Kafka would close the connection 
after one message is sent over.  The client does not detect that and tries to 
send another message again.  That triggers a RST packet.

Any idea why the Kafka broker would close the connection?

Attached you can find the packets between our client and kafka broker.


20:55:40.834543 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [S], seq 
31787730, win 14600, options [mss 1460,nop,nop,sackOK,nop,wscale 9], length 0
0x:  4500 0034 8cc1 4000 4006 ca67 ac12 45c2  E..4..@.@..g..E.
0x0010:  ac12 45b4 868d 2384 01e5 0ad2    ..E...#.
0x0020:  8002 3908 e3c1  0204 05b4 0101 0402  ..9.
0x0030:  0103 0309
20:55:40.834744 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [S.], seq 
1238329644, ack 31787731, win 14600, options [mss 1460,nop,nop,sackOK,nop,wscale 
1], length 0
0x:  4500 0034  4000 4006 5729 ac12 45b4  E..4..@.@.W)..E.
0x0010:  ac12 45c2 2384 868d 49cf 692c 01e5 0ad3  ..E.#...I.i,
0x0020:  8012 3908 e89e  0204 05b4 0101 0402  ..9.
0x0030:  0103 0301
20:55:40.834787 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [.], ack 1, 
win 29, length 0
0x:  4500 0028 8cc2 4000 4006 ca72 ac12 45c2  E..(..@.@..r..E.
0x0010:  ac12 45b4 868d 2384 01e5 0ad3 49cf 692d  ..E...#.I.i-
0x0020:  5010 001d e3b5   P...
20:55:40.834921 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [P.], seq 
1:691, ack 1, win 29, length 690
0x:  4500 02da 8cc3 4000 4006 c7bf ac12 45c2  E.@.@.E.
0x0010:  ac12 45b4 868d 2384 01e5 0ad3 49cf 692d  ..E...#.I.i-
0x0020:  5018 001d e667   02ae    Pg..
0x0030:   0003 000c 6b61 666b 612d 7079 7468  ..kafka-pyth
0x0040:  6f6e 0001  03e8  0001 000e 6576  onev
0x0050:  656e 745f 6e73 706f 6c69 6379  0001  ent_nspolicy
0x0060:     0272      ...r
0x0070:   0266 4ff3 bd11   0004 3131  ...fO.11
0x0080:  3238  0254 5b30 2c7b 2261 7022 3a22  28...T[0,{"ap":"

20:55:40.835297 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [.], ack 
691, win 7990, length 0
0x:  4500 0028 e872 4000 4006 6ec2 ac12 45b4  E..(.r@.@.n...E.
0x0010:  ac12 45c2 2384 868d 49cf 692d 01e5 0d85  ..E.#...I.i-
0x0020:  5010 1f36 408b       P..6@.
20:55:40.837837 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [P.], seq 
1:47, ack 691, win 7990, length 46
0x:  4500 0056 e873 4000 4006 6e93 ac12 45b4  E..V.s@.@.n...E.
0x0010:  ac12 45c2 2384 868d 49cf 692d 01e5 0d85  ..E.#...I.i-
0x0020:  5018 1f36 ece3   002a  0003  P..6...*
0x0030:   0001 000e 6576 656e 745f 6e73 706f  ..event_nspo
0x0040:  6c69 6379  0001      licy
0x0050:   0003 6527   e'
20:55:40.837853 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [.], ack 47, 
win 29, length 0
0x:  4500 0028 8cc4 4000 4006 ca70 ac12 45c2  E..(..@.@..p..E.
0x0010:  ac12 45b4 868d 2384 01e5 0d85 49cf 695b  ..E...#.I.i[
0x0020:  5010 001d e3b5   P...


Closed here

21:05:40.839440 IP 172.18.69.180.9092 > 172.18.69.194.34445: Flags [F.], seq 
47, ack 691, win 7990, length 0
0x:  4500 0028 e874 4000 4006 6ec0 ac12 45b4  E..(.t@.@.n...E.
0x0010:  ac12 45c2 2384 868d 49cf 695b 01e5 0d85  ..E.#...I.i[
0x0020:  5011 1f36 405c       P..6@\
21:05:40.876047 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [.], ack 48, 
win 29, length 0
0x:  4500 0028 8cc5 4000 4006 ca6f ac12 45c2  E..(..@.@..o..E.
0x0010:  ac12 45b4 868d 2384 01e5 0d85 49cf 695c  ..E...#.I.i\
0x0020:  5010 001d e3b5   P...

21:06:17.673188 IP 172.18.69.194.34445 > 172.18.69.180.9092: Flags [P.], seq 
691:1414, ack 48, win 29, length 723
0x:  4500 02fb 8cc6 4000 4006 c79b ac12 45c2  E.@.@.E.
0x0010:  ac12 45b4 868d 2384 01e5 0d85 49cf 695c  ..E...#.I.i\
0x0020:  

log compaction

[Francesco laTorre]

Hi,

We want to enable log compaction on an existing topic (in production).
Is it a safe operation or there are things to take into consideration ?

Kafka version 0.8

Cheers,
Francesco

-- 
 Francesco laTorre
Senior Developer
T: +44 208 742 1600
+44 203 249 8394

E: francesco.lato...@openbet.com
W: www.openbet.com
OpenBet Ltd
Chiswick Park Building 9
566 Chiswick High Rd
London
W4 5XT

This message is confidential and intended only for the addressee. If you
have received this message in error, please immediately notify the
postmas...@openbet.com and delete it from your system as well as any
copies. The content of e-mails as well as traffic data may be monitored by
OpenBet for employment and security purposes. To protect the environment
please do not print this e-mail unless necessary. OpenBet Ltd. Registered
Office: Chiswick Park Building 9, 566 Chiswick High Road, London, W4 5XT,
United Kingdom. A company registered in England and Wales. Registered no.
3134634. VAT no. GB927523612

Re: 0.8.2.1 Client not able to connect with Kafka 0.10.0.1 cluster even the cluster has message format version 0.8.2.

[Madhukar Bharti]

Hi,

After checking we found that there was an issue with version id passed in
ConsumerMetadataRequest after setting it to
0(ConsumerMetadataRequest.currentVersion()). It started working!

Thanks!
Madhukar

On Tue, Nov 1, 2016 at 10:29 PM, Madhukar Bharti 
wrote:

> Hi Ismael,
>
> Below is the stack trace from client.
>
> java.io.EOFException: Received -1 when reading from channel, socket has
>>> likely been closed.
>>
>> at kafka.utils.Utils$.read(Utils.scala:381)
>>
>> at kafka.network.BoundedByteBufferReceive.readFrom(
>>> BoundedByteBufferReceive.scala:54)
>>
>> at kafka.network.Receive$class.readCompletely(Transmission.scala:56)
>>
>> at kafka.network.BoundedByteBufferReceive.readCompletely(
>>> BoundedByteBufferReceive.scala:29)
>>
>> at kafka.network.BlockingChannel.receive(BlockingChannel.scala:111)
>>
>>
> This we are getting while reading from BlockingChannel after sending
> ConsumerMetadataRequest.
> Sample code is here
> 
> .
>
> Regards,
> Madhukar
>
>
> On Tue, Nov 1, 2016 at 9:51 PM, Ismael Juma  wrote:
>
>> OffsetFetchRequest should have api key = 9, but for some reason your
>> broker
>> is receiving a request with api key = 10 (GroupCoordinatorRequest). Can
>> you
>> provide the stacktrace from the client as well?
>>
>> Ismael
>>
>> On Tue, Nov 1, 2016 at 12:13 PM, Madhukar Bharti <
>> bhartimadhu...@gmail.com>
>> wrote:
>>
>> > Hi,
>> >
>> > We have upgraded our cluster from 0.8.2.1 to 0.10.0.1. Now
>> > *log.message.format.version* is set to *0.8.2. *
>> > We are still using Kafka 0.8.2.1 jars in client side, when we tried to
>> > fetch consumer offset using BlockingChannel and OffsetFetchRequest as in
>> > example here
>> > > > Committing+and+fetching+consumer+offsets+in+Kafka>,
>> > we are getting below exceptions in brokers.
>> >
>> > ERROR Closing socket for 172.19.8.200:9092-192.168.1.1:43682 because of
>> > > error (kafka.network.Processor)
>> > > kafka.network.InvalidRequestException: Error getting request for
>> apiKey:
>> > > 10 and apiVersion: 1
>> > > at
>> > > kafka.network.RequestChannel$Request.liftedTree2$1(
>> > RequestChannel.scala:95)
>> > > at
>> > > kafka.network.RequestChannel$Request.(RequestChannel.scala:87)
>> > > at
>> > > kafka.network.Processor$$anonfun$processCompletedReceives$1.
>> > apply(SocketServer.scala:488)
>> > > at
>> > > kafka.network.Processor$$anonfun$processCompletedReceives$1.
>> > apply(SocketServer.scala:483)
>> > > at scala.collection.Iterator$clas
>> s.foreach(Iterator.scala:893)
>> > > at scala.collection.AbstractIterator.foreach(
>> > Iterator.scala:1336)
>> > > at
>> > > scala.collection.IterableLike$class.foreach(IterableLike.scala:72)
>> > > at scala.collection.AbstractItera
>> ble.foreach(Iterable.scala:54)
>> > > at
>> > > kafka.network.Processor.processCompletedReceives(SocketServe
>> r.scala:483)
>> > > at kafka.network.Processor.run(SocketServer.scala:413)
>> > > at java.lang.Thread.run(Thread.java:745)
>> > > Caused by: java.lang.IllegalArgumentException: Invalid version for
>> API
>> > key
>> > > 10: 1
>> > > at
>> > > org.apache.kafka.common.protocol.ProtoUtils.schemaFor(
>> > ProtoUtils.java:31)
>> > > at
>> > > org.apache.kafka.common.protocol.ProtoUtils.
>> > requestSchema(ProtoUtils.java:44)
>> > > at
>> > > org.apache.kafka.common.protocol.ProtoUtils.
>> > parseRequest(ProtoUtils.java:60)
>> > > at
>> > > org.apache.kafka.common.requests.GroupCoordinatorRequest.parse(
>> > GroupCoordinatorRequest.java:59)
>> > > at
>> > > org.apache.kafka.common.requests.AbstractRequest.
>> > getRequest(AbstractRequest.java:54)
>> > > at
>> > > kafka.network.RequestChannel$Request.liftedTree2$1(
>> > RequestChannel.scala:92)
>> > > ... 10 more
>> > >
>> >
>> > But, the same code is working fine with Kafka 0.8.2.1 cluster.
>> > I am aware of some protocol changes has been made in Kafka-0.10.X.X but
>> > don't want to update our client to 0.10.0.1 as of now. Is there any way
>> > without updating client the same code will still give the consumer
>> offset.
>> >
>> >
>> >
>> >
>> > --
>> > Thanks and Regards,
>> > Madhukar Bharti
>> > Mob: 7845755539
>> >
>>
>
>

RE: windowing with the processor api

[Hamza HACHANI]

Hi Eno,


What I want to say is that i don't find a place where to define the size of the 
window and where to precise the time of the advance.


Hamza

Thanks


De : Eno Thereska 
Envoyé : mardi 1 novembre 2016 22:44:47
À : users@kafka.apache.org
Objet : Re: windowing with the processor api

Hi Hamza,

Are you getting a particular error? Here is an example :

Stores.create("window-store")
  .withStringKeys()
  .withStringValues()
  .persistent()
  .windowed(10, 10, 2, false).build(), "the-processor")

Thanks
Eno

> On 2 Nov 2016, at 08:19, Hamza HACHANI  wrote:
>
> Hi,
>
> I would like to know if somebody has an idea how to define the size of the 
> window in the processor api.
>
> I've been blocked for 6 days looking for a solution.
>
> using :
>
> Stores.create(...).withStringKeys().withStringValues().persistent().windowed(...).build()
>
> I was able to define the retention time but not the the size of the window.
>
> Please help me if possible.
>
> Thanks,
>
> Hamza

Re: windowing with the processor api

[Eno Thereska]

Hi Hamza,

Are you getting a particular error? Here is an example :

Stores.create("window-store")
  .withStringKeys()
  .withStringValues()
  .persistent()
  .windowed(10, 10, 2, false).build(), "the-processor")

Thanks
Eno

> On 2 Nov 2016, at 08:19, Hamza HACHANI  wrote:
> 
> Hi,
> 
> I would like to know if somebody has an idea how to define the size of the 
> window in the processor api.
> 
> I've been blocked for 6 days looking for a solution.
> 
> using :
> 
> Stores.create(...).withStringKeys().withStringValues().persistent().windowed(...).build()
> 
> I was able to define the retention time but not the the size of the window.
> 
> Please help me if possible.
> 
> Thanks,
> 
> Hamza

windowing with the processor api

[Hamza HACHANI]

Hi,

I would like to know if somebody has an idea how to define the size of the 
window in the processor api.

I've been blocked for 6 days looking for a solution.

using :

Stores.create(...).withStringKeys().withStringValues().persistent().windowed(...).build()

I was able to define the retention time but not the the size of the window.

Please help me if possible.

Thanks,

Hamza

Re: Question regarding dynamic subscriber environment

[Gerrit Jansen van Vuuren]

Hi,

Have a look at the kafka client lib
https://github.com/gerritjvv/kafka-fast#java-1, it already provides this
functionality.


On Wed, Nov 2, 2016 at 2:34 AM, Janagan Sivagnanasundaram <
janagan1...@gmail.com> wrote:

> Kafka's current nature is does not support to dynamic subscriber
> environment where the topic interest of the subscriber is keep change
> overtime unless subscriber terminate the particular connection and
> reconnect.
>
> What I have planned to implement or design is to enable the dynamic
> subscriber environment where the subscriber can change his topic interests
> without terminating the connection with it's broker. Following are some
> ideas that we can use to do such thing.
>
> 1) Modifying the current protocol by adding an extra attribute
> 2) Create a control connection channel from subscriber to broker
>
> Is there easy/efficient way to do such thing?
>