Re: handling generics in Kafka Scala

Ah, yes. Scala interprets Java collection types, like ArrayList[T], as *invariant*, which means that you can't use a declared ArrayList[String] where you expect an ArrayList[Any], which would be an example of *covariance. *(This is due to Java's flawed way of declaring generics, where the person *d

How to decommission a broker so the controller doesn't return it in the list of known brokers?

How do I permanently remove a broker from a Kafka cluster? Scenario: I have a stable cluster of 3 brokers. I temporarily added a fourth broker that successfully joined the cluster. The controller returned metadata indicating this broker was part of the cluster. However, I never rebalanced partit

RE: handling generics in Kafka Scala

yes that function compiled but everytime I implement createNewConsumer() scala compiler wants to downcast [java.util.ArrayList[Byte]],java.util.ArrayList[Byte]] to wildcard ? consumer = createNewConsumer() //getConsumer() [ERROR] found : org.apache.kafka.clients.consumer.KafkaConsumer[j

RE: handling generics in Kafka Scala

> From: deanwamp...@gmail.com > Date: Wed, 31 Aug 2016 10:53:49 -0500 > Subject: Re: handling generics in Kafka Scala > To: users@kafka.apache.org > > Okay, the type parameters with the variances need to be after the method > name, like this: > > private def createNewConsumer[K <: java.util.Ar

enhancing KStream DSL

Let’s say I have this: KStream[] branches = allRecords .branch( (imsi, callRecord) -> "VOICE".equalsIgnoreCase(callRecord.getCallCommType()), (imsi, callRecord) -> "DATA".equalsIgnoreCase(callRecord.getCallCommType()), (imsi, callRecord) -> true ); KS

Re: KIP-33 Opt out from Time Based indexing

Hi Jun, thanks a lot for the hint, Ill check it out when I get a free minute! Best Jan On 07.09.2016 00:35, Jun Rao wrote: Jan, For the time rolling issue, Jiangjie has committed a fix ( https://issues.apache.org/jira/browse/KAFKA-4099) to trunk. Perhaps you can help test out trunk and see if

Kafka Streams windows strange behavior

Hi, I am trying to implement simple scenario on streams library of kafka. I insert data to kafka topic 1 tuple/second. Streams library is connected to particular topic and what it does is: 1. construct 8 second windows with 4 second sliding time, 2. sum values of tuples (p

org.apache.kafka.common.errors.TimeoutException: Batch Expired

Hello everyone, We have seen this exception couple of times - "org.apache.kafka.common.errors.TimeoutException: Batch Expired". It doesn't happen all the time but mostly during hiccups in network. There is an open jira KAFKA-3686 . I’m planning t