You are viewing a plain text version of this content. The canonical link for it is here.
Posted to users@kafka.apache.org by Debasish Ghosh <gh...@gmail.com> on 2017/07/28 11:55:28 UTC
Kafka Streams state store issue on cluster
Hi -
In my Kafka Streams application, I have a state store resulting from a
stateful streaming topology. The environment is
- Kafka 0.10.2.1
- It runs on a DC/OS cluster
- I am running Confluent-Kafka 3.2.2 on the cluster
- Each topic that I have has 2 partitions with replication factor = 2
- The application also has an associated http service that does
interactive queries on the state store
The application runs fine when I invoke a single instance. I can use the
http endpoints to do queries and everything looks good. Problems surface
when I try to spawn another instance of the application. I use the same
APPLICATION_ID_CONFIG and the instance starts on a different node of the
cluster. The data consumption part works fine as the new instance also
starts consuming from the same topic as the first one. But when I try the
http query, the metadata fetch fails ..
I have some code snippet like this as part of the query that tries to fetch
the metadata so that I can locate the host to query on ..
metadataService.streamsMetadataForStoreAndKey(store, hostKey,
stringSerializer) match {
case Success(host) => {
// hostKey is on another instance. call the other instance to fetch
the data.
if (!thisHost(host)) {
logger.warn(s"Key $hostKey is on another instance not on $host -
requerying ..")
httpRequester.queryFromHost[Long](host, path)
} else {
// hostKey is on this instance
localStateStoreQuery.queryStateStore(streams, store, hostKey)
}
}
case Failure(ex) => Future.failed(ex)
}
and the metadataService.streamsMetadataForStoreAndKey has the following
call ..
streams.metadataForKey(store, key, serializer) match {
case null => throw new IllegalArgumentException(s"Metadata for key
$key not found in $store")
case metadata => new HostStoreInfo(metadata.host, metadata.port,
metadata.stateStoreNames.asScala.toSet)
}
When I start the second instance, streams.metadataForKey returns null for
any key I pass .. Here's the relevant stack trace ..
java.lang.IllegalArgumentException: Metadata for key mtc.clark.net not
found in access-count-per-host
at
com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
at scala.util.Try$.apply(Try.scala:209)
at
com.xx.fdp.sample.kstream.services.MetadataService.streamsMetadataForStoreAndKey(MetadataService.scala:46)
at
com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryInfo(KeyValueFetcher.scala:36)
at
com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCountSummary(KeyValueFetcher.scala:29)
at
com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun$routes$10(WeblogDSLHttpService.scala:41)
at
akka.http.scaladsl.server.directives.RouteDirectives.$anonfun$complete$1(RouteDirectives.scala:47)
at
akka.http.scaladsl.server.StandardRoute$$anon$1.apply(StandardRoute.scala:19)
at
akka.http.scaladsl.server.StandardRoute$$anon$1.apply(StandardRoute.scala:19)
...
and following this exception I get another one which looks like an internal
exception that stops the application ..
09:57:33.731 TKD [StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread -
stream-thread [StreamThread-1] Streams application error during processing:
java.lang.IllegalStateException: Attempt to retrieve exception from future
which hasn't failed
at
org.apache.kafka.clients.consumer.internals.RequestFuture.exception(RequestFuture.java:99)
at
org.apache.kafka.clients.consumer.internals.RequestFuture.isRetriable(RequestFuture.java:89)
at
org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:590)
at
org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1124)
at
org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:296)
at
org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:79)
at
org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
at
org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:280)
at
org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:807)
at
org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:794)
at
org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:769)
at
org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:647)
at
org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
BTW the application runs fine when I have 2 instances running on the same
host (my laptop) or on 1 instance on the cluster. The problem surfaces only
when I start the second instance.
Any help / pointer / area to look into ?
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: Kafka Streams state store issue on cluster
Posted by Debasish Ghosh <gh...@gmail.com>.
I fixed this problem by decoupling the 2 issues ..
- since I am using an http service for interactive queries, I need an
http interface. For that I was using 0.0.0.0 and a port number. This is
still the same - the only change I made is to have a Mesos configuration
where Marathon picks up a random port number for me. This ensures that I
don't ever end up with a port that's currently being used by something else.
- for APPLICATION_SERVER_CONFIG, I cannot use 0.0.0.0 since I need an
actual address so that I can use it to query Kafka streams metadata and
issue a subsequent http GET to fetch the actual data from the host that has
it. So when setting APPLICATION_SERVER_CONFIG I use the actual INetAddress
of the host.
This fixes the problem.
regards.
On Fri, Jul 28, 2017 at 8:46 PM, Damian Guy <da...@gmail.com> wrote:
> Hmmm, i'm not sure that is going to work as both nodes will have the same
> setting for StreamsConfig.APPLICATION_SERVER_PORT, i.e, 0.0.0.0:7070
>
> On Fri, 28 Jul 2017 at 16:02 Debasish Ghosh <gh...@gmail.com>
> wrote:
>
>> The log file is a huge one. I can send it to you though. Before that let
>> me confirm one point ..
>>
>> I set the APPLICATION_SERVER_CONFIG to s"${config.httpInterface}:${
>> config.httpPort}". In my case the httpInterface is "0.0.0.0" and the port
>> is set to 7070. Since the two instances start on different nodes, this
>> should be ok - right ?
>>
>> regards.
>>
>> On Fri, Jul 28, 2017 at 8:18 PM, Damian Guy <da...@gmail.com> wrote:
>>
>>> Do you have any logs that might help to work out what is going wrong?
>>>
>>> On Fri, 28 Jul 2017 at 14:16 Damian Guy <da...@gmail.com> wrote:
>>>
>>>> The config looks ok to me
>>>>
>>>> On Fri, 28 Jul 2017 at 13:24 Debasish Ghosh <gh...@gmail.com>
>>>> wrote:
>>>>
>>>>> I am setting APPLICATION_SERVER_CONFIG, which is possibly what u r
>>>>> referring to. Just now I noticed that I may also need to set
>>>>> REPLICATION_FACTOR_CONFIG, which needs to be set to 2 (default is 1).
>>>>> Anything else that I may be missing ?
>>>>>
>>>>>
>>>>> regards.
>>>>>
>>>>> On Fri, Jul 28, 2017 at 5:46 PM, Debasish Ghosh <
>>>>> ghosh.debasish@gmail.com>
>>>>> wrote:
>>>>>
>>>>> > Hi Damien -
>>>>> >
>>>>> > I am not sure I understand what u mean .. I have the following set
>>>>> in the
>>>>> > application .. Do I need to set anything else at the host level ?
>>>>> > Environment variable ?
>>>>> >
>>>>> > val streamingConfig = {
>>>>> > val settings = new Properties
>>>>> > settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
>>>>> > "kstream-weblog-processing")
>>>>> > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
>>>>> config.brokers)
>>>>> >
>>>>> > config.schemaRegistryUrl.foreach{ url =>
>>>>> > settings.put(AbstractKafkaAvroSerDeConfig.
>>>>> SCHEMA_REGISTRY_URL_CONFIG,
>>>>> > url)
>>>>> > }
>>>>> >
>>>>> > settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
>>>>> > Serdes.ByteArray.getClass.getName)
>>>>> > settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
>>>>> > Serdes.String.getClass.getName)
>>>>> >
>>>>> > // setting offset reset to earliest so that we can re-run the
>>>>> demo
>>>>> > code with the same pre-loaded data
>>>>> > // Note: To re-run the demo, you need to use the offset reset
>>>>> tool:
>>>>> > // https://cwiki.apache.org/confluence/display/KAFKA/
>>>>> > Kafka+Streams+Application+Reset+Tool
>>>>> > settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
>>>>> "earliest")
>>>>> >
>>>>> > // need this for query service
>>>>> > settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
>>>>> > s"${config.httpInterface}:${config.httpPort}")
>>>>> >
>>>>> > // default is /tmp/kafka-streams
>>>>> > settings.put(StreamsConfig.STATE_DIR_CONFIG,
>>>>> config.stateStoreDir)
>>>>> >
>>>>> > // Set the commit interval to 500ms so that any changes are
>>>>> flushed
>>>>> > frequently and the summary
>>>>> > // data are updated with low latency.
>>>>> > settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
>>>>> >
>>>>> > settings
>>>>> > }
>>>>> >
>>>>> > Please explain a bit ..
>>>>> >
>>>>> > regards.
>>>>> >
>>>>> >
>>>>> > On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com>
>>>>> wrote:
>>>>> >
>>>>> >> Hi,
>>>>> >>
>>>>> >> Do you have the application.server property set appropriately for
>>>>> both
>>>>> >> hosts?
>>>>> >>
>>>>> >> The second stack trace is this bug:
>>>>> >> https://issues.apache.org/jira/browse/KAFKA-5556
>>>>> >>
>>>>> >> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <
>>>>> ghosh.debasish@gmail.com>
>>>>> >> wrote:
>>>>> >>
>>>>> >> > Hi -
>>>>> >> >
>>>>> >> > In my Kafka Streams application, I have a state store resulting
>>>>> from a
>>>>> >> > stateful streaming topology. The environment is
>>>>> >> >
>>>>> >> > - Kafka 0.10.2.1
>>>>> >> > - It runs on a DC/OS cluster
>>>>> >> > - I am running Confluent-Kafka 3.2.2 on the cluster
>>>>> >> > - Each topic that I have has 2 partitions with replication
>>>>> factor = 2
>>>>> >> > - The application also has an associated http service that does
>>>>> >> > interactive queries on the state store
>>>>> >> >
>>>>> >> > The application runs fine when I invoke a single instance. I can
>>>>> use the
>>>>> >> > http endpoints to do queries and everything looks good. Problems
>>>>> surface
>>>>> >> > when I try to spawn another instance of the application. I use
>>>>> the same
>>>>> >> > APPLICATION_ID_CONFIG and the instance starts on a different node
>>>>> of the
>>>>> >> > cluster. The data consumption part works fine as the new instance
>>>>> also
>>>>> >> > starts consuming from the same topic as the first one. But when I
>>>>> try
>>>>> >> the
>>>>> >> > http query, the metadata fetch fails ..
>>>>> >> >
>>>>> >> > I have some code snippet like this as part of the query that
>>>>> tries to
>>>>> >> fetch
>>>>> >> > the metadata so that I can locate the host to query on ..
>>>>> >> >
>>>>> >> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
>>>>> >> > stringSerializer) match {
>>>>> >> > case Success(host) => {
>>>>> >> > // hostKey is on another instance. call the other
>>>>> instance to
>>>>> >> fetch
>>>>> >> > the data.
>>>>> >> > if (!thisHost(host)) {
>>>>> >> > logger.warn(s"Key $hostKey is on another instance not on
>>>>> >> $host -
>>>>> >> > requerying ..")
>>>>> >> > httpRequester.queryFromHost[Long](host, path)
>>>>> >> > } else {
>>>>> >> > // hostKey is on this instance
>>>>> >> > localStateStoreQuery.queryStateStore(streams, store,
>>>>> hostKey)
>>>>> >> > }
>>>>> >> > }
>>>>> >> > case Failure(ex) => Future.failed(ex)
>>>>> >> > }
>>>>> >> >
>>>>> >> > and the metadataService.streamsMetadataForStoreAndKey has the
>>>>> following
>>>>> >> > call ..
>>>>> >> >
>>>>> >> > streams.metadataForKey(store, key, serializer) match {
>>>>> >> > case null => throw new IllegalArgumentException(s"Metadata
>>>>> for
>>>>> >> key
>>>>> >> > $key not found in $store")
>>>>> >> > case metadata => new HostStoreInfo(metadata.host,
>>>>> metadata.port,
>>>>> >> > metadata.stateStoreNames.asScala.toSet)
>>>>> >> > }
>>>>> >> >
>>>>> >> > When I start the second instance, streams.metadataForKey returns
>>>>> null
>>>>> >> for
>>>>> >> > any key I pass .. Here's the relevant stack trace ..
>>>>> >> >
>>>>> >> > java.lang.IllegalArgumentException: Metadata for key
>>>>> mtc.clark.net not
>>>>> >> > found in access-count-per-host
>>>>> >> > at
>>>>> >> >
>>>>> >> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
>>>>> >> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
>>>>> >> > at scala.util.Try$.apply(Try.scala:209)
>>>>> >> > at
>>>>> >> >
>>>>> >> > com.xx.fdp.sample.kstream.services.MetadataService.streamsMe
>>>>> >> tadataForStoreAndKey(MetadataService.scala:46)
>>>>> >> > at
>>>>> >> >
>>>>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryI
>>>>> >> nfo(KeyValueFetcher.scala:36)
>>>>> >> > at
>>>>> >> >
>>>>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCo
>>>>> >> untSummary(KeyValueFetcher.scala:29)
>>>>> >> > at
>>>>> >> >
>>>>> >> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun
>>>>> >> $routes$10(WeblogDSLHttpService.scala:41)
>>>>> >> > at
>>>>> >> >
>>>>> >> > akka.http.scaladsl.server.directives.RouteDirectives.$anonfu
>>>>> >> n$complete$1(RouteDirectives.scala:47)
>>>>> >> > at
>>>>> >> >
>>>>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>>>>> >> rdRoute.scala:19)
>>>>> >> > at
>>>>> >> >
>>>>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>>>>> >> rdRoute.scala:19)
>>>>> >> > ...
>>>>> >> >
>>>>> >> > and following this exception I get another one which looks like an
>>>>> >> internal
>>>>> >> > exception that stops the application ..
>>>>> >> >
>>>>> >> > 09:57:33.731 TKD [StreamThread-1] ERROR o.a.k.s.p.internals.
>>>>> StreamThread
>>>>> >> -
>>>>> >> > stream-thread [StreamThread-1] Streams application error during
>>>>> >> processing:
>>>>> >> > java.lang.IllegalStateException: Attempt to retrieve exception
>>>>> from
>>>>> >> future
>>>>> >> > which hasn't failed
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>>>>> >> exception(RequestFuture.java:99)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>>>>> >> isRetriable(RequestFuture.java:89)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.clients.consumer.internals.ConsumerCoordina
>>>>> >> tor.commitOffsetsSync(ConsumerCoordinator.java:590)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(
>>>>> >> KafkaConsumer.java:1124)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>>>>> >> commitOffsets(StreamTask.java:296)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamTask$1.
>>>>> >> run(StreamTask.java:79)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamsMetricsI
>>>>> >> mpl.measureLatencyNs(StreamsMetricsImpl.java:188)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>>>>> >> commit(StreamTask.java:280)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>>> >> commitOne(StreamThread.java:807)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>>> >> commitAll(StreamThread.java:794)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>>> >> maybeCommit(StreamThread.java:769)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>>> >> runLoop(StreamThread.java:647)
>>>>> >> > at
>>>>> >> >
>>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>>> >> run(StreamThread.java:361)
>>>>> >> >
>>>>> >> > BTW the application runs fine when I have 2 instances running on
>>>>> the
>>>>> >> same
>>>>> >> > host (my laptop) or on 1 instance on the cluster. The problem
>>>>> surfaces
>>>>> >> only
>>>>> >> > when I start the second instance.
>>>>> >> >
>>>>> >> > Any help / pointer / area to look into ?
>>>>> >> >
>>>>> >> > --
>>>>> >> > Debasish Ghosh
>>>>> >> > http://manning.com/ghosh2
>>>>> >> > http://manning.com/ghosh
>>>>> >> >
>>>>> >> > Twttr: @debasishg
>>>>> >> > Blog: http://debasishg.blogspot.com
>>>>> >> > Code: http://github.com/debasishg
>>>>> >> >
>>>>> >>
>>>>> >
>>>>> >
>>>>> >
>>>>> > --
>>>>> > Debasish Ghosh
>>>>> > http://manning.com/ghosh2
>>>>> > http://manning.com/ghosh
>>>>> >
>>>>> > Twttr: @debasishg
>>>>> > Blog: http://debasishg.blogspot.com
>>>>> > Code: http://github.com/debasishg
>>>>> >
>>>>>
>>>>>
>>>>>
>>>>> --
>>>>> Debasish Ghosh
>>>>> http://manning.com/ghosh2
>>>>> http://manning.com/ghosh
>>>>>
>>>>> Twttr: @debasishg
>>>>> Blog: http://debasishg.blogspot.com
>>>>> Code: http://github.com/debasishg
>>>>>
>>>>
>>
>>
>> --
>> Debasish Ghosh
>> http://manning.com/ghosh2
>> http://manning.com/ghosh
>>
>> Twttr: @debasishg
>> Blog: http://debasishg.blogspot.com
>> Code: http://github.com/debasishg
>>
>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: Kafka Streams state store issue on cluster
Posted by Damian Guy <da...@gmail.com>.
Hmmm, i'm not sure that is going to work as both nodes will have the same
setting for StreamsConfig.APPLICATION_SERVER_PORT, i.e, 0.0.0.0:7070
On Fri, 28 Jul 2017 at 16:02 Debasish Ghosh <gh...@gmail.com>
wrote:
> The log file is a huge one. I can send it to you though. Before that let
> me confirm one point ..
>
> I set the APPLICATION_SERVER_CONFIG to
> s"${config.httpInterface}:${config.httpPort}". In my case the
> httpInterface is "0.0.0.0" and the port is set to 7070. Since the two
> instances start on different nodes, this should be ok - right ?
>
> regards.
>
> On Fri, Jul 28, 2017 at 8:18 PM, Damian Guy <da...@gmail.com> wrote:
>
>> Do you have any logs that might help to work out what is going wrong?
>>
>> On Fri, 28 Jul 2017 at 14:16 Damian Guy <da...@gmail.com> wrote:
>>
>>> The config looks ok to me
>>>
>>> On Fri, 28 Jul 2017 at 13:24 Debasish Ghosh <gh...@gmail.com>
>>> wrote:
>>>
>>>> I am setting APPLICATION_SERVER_CONFIG, which is possibly what u r
>>>> referring to. Just now I noticed that I may also need to set
>>>> REPLICATION_FACTOR_CONFIG, which needs to be set to 2 (default is 1).
>>>> Anything else that I may be missing ?
>>>>
>>>>
>>>> regards.
>>>>
>>>> On Fri, Jul 28, 2017 at 5:46 PM, Debasish Ghosh <
>>>> ghosh.debasish@gmail.com>
>>>> wrote:
>>>>
>>>> > Hi Damien -
>>>> >
>>>> > I am not sure I understand what u mean .. I have the following set in
>>>> the
>>>> > application .. Do I need to set anything else at the host level ?
>>>> > Environment variable ?
>>>> >
>>>> > val streamingConfig = {
>>>> > val settings = new Properties
>>>> > settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
>>>> > "kstream-weblog-processing")
>>>> > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
>>>> config.brokers)
>>>> >
>>>> > config.schemaRegistryUrl.foreach{ url =>
>>>> >
>>>> settings.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
>>>> > url)
>>>> > }
>>>> >
>>>> > settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
>>>> > Serdes.ByteArray.getClass.getName)
>>>> > settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
>>>> > Serdes.String.getClass.getName)
>>>> >
>>>> > // setting offset reset to earliest so that we can re-run the
>>>> demo
>>>> > code with the same pre-loaded data
>>>> > // Note: To re-run the demo, you need to use the offset reset
>>>> tool:
>>>> > // https://cwiki.apache.org/confluence/display/KAFKA/
>>>> > Kafka+Streams+Application+Reset+Tool
>>>> > settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
>>>> "earliest")
>>>> >
>>>> > // need this for query service
>>>> > settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
>>>> > s"${config.httpInterface}:${config.httpPort}")
>>>> >
>>>> > // default is /tmp/kafka-streams
>>>> > settings.put(StreamsConfig.STATE_DIR_CONFIG,
>>>> config.stateStoreDir)
>>>> >
>>>> > // Set the commit interval to 500ms so that any changes are
>>>> flushed
>>>> > frequently and the summary
>>>> > // data are updated with low latency.
>>>> > settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
>>>> >
>>>> > settings
>>>> > }
>>>> >
>>>> > Please explain a bit ..
>>>> >
>>>> > regards.
>>>> >
>>>> >
>>>> > On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com>
>>>> wrote:
>>>> >
>>>> >> Hi,
>>>> >>
>>>> >> Do you have the application.server property set appropriately for
>>>> both
>>>> >> hosts?
>>>> >>
>>>> >> The second stack trace is this bug:
>>>> >> https://issues.apache.org/jira/browse/KAFKA-5556
>>>> >>
>>>> >> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <
>>>> ghosh.debasish@gmail.com>
>>>> >> wrote:
>>>> >>
>>>> >> > Hi -
>>>> >> >
>>>> >> > In my Kafka Streams application, I have a state store resulting
>>>> from a
>>>> >> > stateful streaming topology. The environment is
>>>> >> >
>>>> >> > - Kafka 0.10.2.1
>>>> >> > - It runs on a DC/OS cluster
>>>> >> > - I am running Confluent-Kafka 3.2.2 on the cluster
>>>> >> > - Each topic that I have has 2 partitions with replication
>>>> factor = 2
>>>> >> > - The application also has an associated http service that does
>>>> >> > interactive queries on the state store
>>>> >> >
>>>> >> > The application runs fine when I invoke a single instance. I can
>>>> use the
>>>> >> > http endpoints to do queries and everything looks good. Problems
>>>> surface
>>>> >> > when I try to spawn another instance of the application. I use the
>>>> same
>>>> >> > APPLICATION_ID_CONFIG and the instance starts on a different node
>>>> of the
>>>> >> > cluster. The data consumption part works fine as the new instance
>>>> also
>>>> >> > starts consuming from the same topic as the first one. But when I
>>>> try
>>>> >> the
>>>> >> > http query, the metadata fetch fails ..
>>>> >> >
>>>> >> > I have some code snippet like this as part of the query that tries
>>>> to
>>>> >> fetch
>>>> >> > the metadata so that I can locate the host to query on ..
>>>> >> >
>>>> >> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
>>>> >> > stringSerializer) match {
>>>> >> > case Success(host) => {
>>>> >> > // hostKey is on another instance. call the other instance
>>>> to
>>>> >> fetch
>>>> >> > the data.
>>>> >> > if (!thisHost(host)) {
>>>> >> > logger.warn(s"Key $hostKey is on another instance not on
>>>> >> $host -
>>>> >> > requerying ..")
>>>> >> > httpRequester.queryFromHost[Long](host, path)
>>>> >> > } else {
>>>> >> > // hostKey is on this instance
>>>> >> > localStateStoreQuery.queryStateStore(streams, store,
>>>> hostKey)
>>>> >> > }
>>>> >> > }
>>>> >> > case Failure(ex) => Future.failed(ex)
>>>> >> > }
>>>> >> >
>>>> >> > and the metadataService.streamsMetadataForStoreAndKey has the
>>>> following
>>>> >> > call ..
>>>> >> >
>>>> >> > streams.metadataForKey(store, key, serializer) match {
>>>> >> > case null => throw new IllegalArgumentException(s"Metadata
>>>> for
>>>> >> key
>>>> >> > $key not found in $store")
>>>> >> > case metadata => new HostStoreInfo(metadata.host,
>>>> metadata.port,
>>>> >> > metadata.stateStoreNames.asScala.toSet)
>>>> >> > }
>>>> >> >
>>>> >> > When I start the second instance, streams.metadataForKey returns
>>>> null
>>>> >> for
>>>> >> > any key I pass .. Here's the relevant stack trace ..
>>>> >> >
>>>> >> > java.lang.IllegalArgumentException: Metadata for key mtc.clark.net
>>>> not
>>>> >> > found in access-count-per-host
>>>> >> > at
>>>> >> >
>>>> >> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
>>>> >> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
>>>> >> > at scala.util.Try$.apply(Try.scala:209)
>>>> >> > at
>>>> >> >
>>>> >> > com.xx.fdp.sample.kstream.services.MetadataService.streamsMe
>>>> >> tadataForStoreAndKey(MetadataService.scala:46)
>>>> >> > at
>>>> >> >
>>>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryI
>>>> >> nfo(KeyValueFetcher.scala:36)
>>>> >> > at
>>>> >> >
>>>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCo
>>>> >> untSummary(KeyValueFetcher.scala:29)
>>>> >> > at
>>>> >> >
>>>> >> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun
>>>> >> $routes$10(WeblogDSLHttpService.scala:41)
>>>> >> > at
>>>> >> >
>>>> >> > akka.http.scaladsl.server.directives.RouteDirectives.$anonfu
>>>> >> n$complete$1(RouteDirectives.scala:47)
>>>> >> > at
>>>> >> >
>>>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>>>> >> rdRoute.scala:19)
>>>> >> > at
>>>> >> >
>>>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>>>> >> rdRoute.scala:19)
>>>> >> > ...
>>>> >> >
>>>> >> > and following this exception I get another one which looks like an
>>>> >> internal
>>>> >> > exception that stops the application ..
>>>> >> >
>>>> >> > 09:57:33.731 TKD [StreamThread-1] ERROR
>>>> o.a.k.s.p.internals.StreamThread
>>>> >> -
>>>> >> > stream-thread [StreamThread-1] Streams application error during
>>>> >> processing:
>>>> >> > java.lang.IllegalStateException: Attempt to retrieve exception from
>>>> >> future
>>>> >> > which hasn't failed
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>>>> >> exception(RequestFuture.java:99)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>>>> >> isRetriable(RequestFuture.java:89)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.clients.consumer.internals.ConsumerCoordina
>>>> >> tor.commitOffsetsSync(ConsumerCoordinator.java:590)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(
>>>> >> KafkaConsumer.java:1124)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>>>> >> commitOffsets(StreamTask.java:296)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamTask$1.
>>>> >> run(StreamTask.java:79)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamsMetricsI
>>>> >> mpl.measureLatencyNs(StreamsMetricsImpl.java:188)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>>>> >> commit(StreamTask.java:280)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>> >> commitOne(StreamThread.java:807)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>> >> commitAll(StreamThread.java:794)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>> >> maybeCommit(StreamThread.java:769)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>> >> runLoop(StreamThread.java:647)
>>>> >> > at
>>>> >> >
>>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>>> >> run(StreamThread.java:361)
>>>> >> >
>>>> >> > BTW the application runs fine when I have 2 instances running on
>>>> the
>>>> >> same
>>>> >> > host (my laptop) or on 1 instance on the cluster. The problem
>>>> surfaces
>>>> >> only
>>>> >> > when I start the second instance.
>>>> >> >
>>>> >> > Any help / pointer / area to look into ?
>>>> >> >
>>>> >> > --
>>>> >> > Debasish Ghosh
>>>> >> > http://manning.com/ghosh2
>>>> >> > http://manning.com/ghosh
>>>> >> >
>>>> >> > Twttr: @debasishg
>>>> >> > Blog: http://debasishg.blogspot.com
>>>> >> > Code: http://github.com/debasishg
>>>> >> >
>>>> >>
>>>> >
>>>> >
>>>> >
>>>> > --
>>>> > Debasish Ghosh
>>>> > http://manning.com/ghosh2
>>>> > http://manning.com/ghosh
>>>> >
>>>> > Twttr: @debasishg
>>>> > Blog: http://debasishg.blogspot.com
>>>> > Code: http://github.com/debasishg
>>>> >
>>>>
>>>>
>>>>
>>>> --
>>>> Debasish Ghosh
>>>> http://manning.com/ghosh2
>>>> http://manning.com/ghosh
>>>>
>>>> Twttr: @debasishg
>>>> Blog: http://debasishg.blogspot.com
>>>> Code: http://github.com/debasishg
>>>>
>>>
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>
Re: Kafka Streams state store issue on cluster
Posted by Debasish Ghosh <gh...@gmail.com>.
The log file is a huge one. I can send it to you though. Before that let me
confirm one point ..
I set the APPLICATION_SERVER_CONFIG to
s"${config.httpInterface}:${config.httpPort}". In my case the httpInterface
is "0.0.0.0" and the port is set to 7070. Since the two instances start on
different nodes, this should be ok - right ?
regards.
On Fri, Jul 28, 2017 at 8:18 PM, Damian Guy <da...@gmail.com> wrote:
> Do you have any logs that might help to work out what is going wrong?
>
> On Fri, 28 Jul 2017 at 14:16 Damian Guy <da...@gmail.com> wrote:
>
>> The config looks ok to me
>>
>> On Fri, 28 Jul 2017 at 13:24 Debasish Ghosh <gh...@gmail.com>
>> wrote:
>>
>>> I am setting APPLICATION_SERVER_CONFIG, which is possibly what u r
>>> referring to. Just now I noticed that I may also need to set
>>> REPLICATION_FACTOR_CONFIG, which needs to be set to 2 (default is 1).
>>> Anything else that I may be missing ?
>>>
>>>
>>> regards.
>>>
>>> On Fri, Jul 28, 2017 at 5:46 PM, Debasish Ghosh <
>>> ghosh.debasish@gmail.com>
>>> wrote:
>>>
>>> > Hi Damien -
>>> >
>>> > I am not sure I understand what u mean .. I have the following set in
>>> the
>>> > application .. Do I need to set anything else at the host level ?
>>> > Environment variable ?
>>> >
>>> > val streamingConfig = {
>>> > val settings = new Properties
>>> > settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
>>> > "kstream-weblog-processing")
>>> > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
>>> config.brokers)
>>> >
>>> > config.schemaRegistryUrl.foreach{ url =>
>>> > settings.put(AbstractKafkaAvroSerDeConfig.
>>> SCHEMA_REGISTRY_URL_CONFIG,
>>> > url)
>>> > }
>>> >
>>> > settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
>>> > Serdes.ByteArray.getClass.getName)
>>> > settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
>>> > Serdes.String.getClass.getName)
>>> >
>>> > // setting offset reset to earliest so that we can re-run the
>>> demo
>>> > code with the same pre-loaded data
>>> > // Note: To re-run the demo, you need to use the offset reset
>>> tool:
>>> > // https://cwiki.apache.org/confluence/display/KAFKA/
>>> > Kafka+Streams+Application+Reset+Tool
>>> > settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG,
>>> "earliest")
>>> >
>>> > // need this for query service
>>> > settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
>>> > s"${config.httpInterface}:${config.httpPort}")
>>> >
>>> > // default is /tmp/kafka-streams
>>> > settings.put(StreamsConfig.STATE_DIR_CONFIG,
>>> config.stateStoreDir)
>>> >
>>> > // Set the commit interval to 500ms so that any changes are
>>> flushed
>>> > frequently and the summary
>>> > // data are updated with low latency.
>>> > settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
>>> >
>>> > settings
>>> > }
>>> >
>>> > Please explain a bit ..
>>> >
>>> > regards.
>>> >
>>> >
>>> > On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com>
>>> wrote:
>>> >
>>> >> Hi,
>>> >>
>>> >> Do you have the application.server property set appropriately for both
>>> >> hosts?
>>> >>
>>> >> The second stack trace is this bug:
>>> >> https://issues.apache.org/jira/browse/KAFKA-5556
>>> >>
>>> >> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <ghosh.debasish@gmail.com
>>> >
>>> >> wrote:
>>> >>
>>> >> > Hi -
>>> >> >
>>> >> > In my Kafka Streams application, I have a state store resulting
>>> from a
>>> >> > stateful streaming topology. The environment is
>>> >> >
>>> >> > - Kafka 0.10.2.1
>>> >> > - It runs on a DC/OS cluster
>>> >> > - I am running Confluent-Kafka 3.2.2 on the cluster
>>> >> > - Each topic that I have has 2 partitions with replication
>>> factor = 2
>>> >> > - The application also has an associated http service that does
>>> >> > interactive queries on the state store
>>> >> >
>>> >> > The application runs fine when I invoke a single instance. I can
>>> use the
>>> >> > http endpoints to do queries and everything looks good. Problems
>>> surface
>>> >> > when I try to spawn another instance of the application. I use the
>>> same
>>> >> > APPLICATION_ID_CONFIG and the instance starts on a different node
>>> of the
>>> >> > cluster. The data consumption part works fine as the new instance
>>> also
>>> >> > starts consuming from the same topic as the first one. But when I
>>> try
>>> >> the
>>> >> > http query, the metadata fetch fails ..
>>> >> >
>>> >> > I have some code snippet like this as part of the query that tries
>>> to
>>> >> fetch
>>> >> > the metadata so that I can locate the host to query on ..
>>> >> >
>>> >> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
>>> >> > stringSerializer) match {
>>> >> > case Success(host) => {
>>> >> > // hostKey is on another instance. call the other instance
>>> to
>>> >> fetch
>>> >> > the data.
>>> >> > if (!thisHost(host)) {
>>> >> > logger.warn(s"Key $hostKey is on another instance not on
>>> >> $host -
>>> >> > requerying ..")
>>> >> > httpRequester.queryFromHost[Long](host, path)
>>> >> > } else {
>>> >> > // hostKey is on this instance
>>> >> > localStateStoreQuery.queryStateStore(streams, store,
>>> hostKey)
>>> >> > }
>>> >> > }
>>> >> > case Failure(ex) => Future.failed(ex)
>>> >> > }
>>> >> >
>>> >> > and the metadataService.streamsMetadataForStoreAndKey has the
>>> following
>>> >> > call ..
>>> >> >
>>> >> > streams.metadataForKey(store, key, serializer) match {
>>> >> > case null => throw new IllegalArgumentException(s"Metadata
>>> for
>>> >> key
>>> >> > $key not found in $store")
>>> >> > case metadata => new HostStoreInfo(metadata.host,
>>> metadata.port,
>>> >> > metadata.stateStoreNames.asScala.toSet)
>>> >> > }
>>> >> >
>>> >> > When I start the second instance, streams.metadataForKey returns
>>> null
>>> >> for
>>> >> > any key I pass .. Here's the relevant stack trace ..
>>> >> >
>>> >> > java.lang.IllegalArgumentException: Metadata for key mtc.clark.net
>>> not
>>> >> > found in access-count-per-host
>>> >> > at
>>> >> >
>>> >> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
>>> >> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
>>> >> > at scala.util.Try$.apply(Try.scala:209)
>>> >> > at
>>> >> >
>>> >> > com.xx.fdp.sample.kstream.services.MetadataService.streamsMe
>>> >> tadataForStoreAndKey(MetadataService.scala:46)
>>> >> > at
>>> >> >
>>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryI
>>> >> nfo(KeyValueFetcher.scala:36)
>>> >> > at
>>> >> >
>>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCo
>>> >> untSummary(KeyValueFetcher.scala:29)
>>> >> > at
>>> >> >
>>> >> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun
>>> >> $routes$10(WeblogDSLHttpService.scala:41)
>>> >> > at
>>> >> >
>>> >> > akka.http.scaladsl.server.directives.RouteDirectives.$anonfu
>>> >> n$complete$1(RouteDirectives.scala:47)
>>> >> > at
>>> >> >
>>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>>> >> rdRoute.scala:19)
>>> >> > at
>>> >> >
>>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>>> >> rdRoute.scala:19)
>>> >> > ...
>>> >> >
>>> >> > and following this exception I get another one which looks like an
>>> >> internal
>>> >> > exception that stops the application ..
>>> >> >
>>> >> > 09:57:33.731 TKD [StreamThread-1] ERROR o.a.k.s.p.internals.
>>> StreamThread
>>> >> -
>>> >> > stream-thread [StreamThread-1] Streams application error during
>>> >> processing:
>>> >> > java.lang.IllegalStateException: Attempt to retrieve exception from
>>> >> future
>>> >> > which hasn't failed
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>>> >> exception(RequestFuture.java:99)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>>> >> isRetriable(RequestFuture.java:89)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.clients.consumer.internals.ConsumerCoordina
>>> >> tor.commitOffsetsSync(ConsumerCoordinator.java:590)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(
>>> >> KafkaConsumer.java:1124)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>>> >> commitOffsets(StreamTask.java:296)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamTask$1.
>>> >> run(StreamTask.java:79)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamsMetricsI
>>> >> mpl.measureLatencyNs(StreamsMetricsImpl.java:188)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>>> >> commit(StreamTask.java:280)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>> >> commitOne(StreamThread.java:807)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>> >> commitAll(StreamThread.java:794)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>> >> maybeCommit(StreamThread.java:769)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>> >> runLoop(StreamThread.java:647)
>>> >> > at
>>> >> >
>>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>>> >> run(StreamThread.java:361)
>>> >> >
>>> >> > BTW the application runs fine when I have 2 instances running on the
>>> >> same
>>> >> > host (my laptop) or on 1 instance on the cluster. The problem
>>> surfaces
>>> >> only
>>> >> > when I start the second instance.
>>> >> >
>>> >> > Any help / pointer / area to look into ?
>>> >> >
>>> >> > --
>>> >> > Debasish Ghosh
>>> >> > http://manning.com/ghosh2
>>> >> > http://manning.com/ghosh
>>> >> >
>>> >> > Twttr: @debasishg
>>> >> > Blog: http://debasishg.blogspot.com
>>> >> > Code: http://github.com/debasishg
>>> >> >
>>> >>
>>> >
>>> >
>>> >
>>> > --
>>> > Debasish Ghosh
>>> > http://manning.com/ghosh2
>>> > http://manning.com/ghosh
>>> >
>>> > Twttr: @debasishg
>>> > Blog: http://debasishg.blogspot.com
>>> > Code: http://github.com/debasishg
>>> >
>>>
>>>
>>>
>>> --
>>> Debasish Ghosh
>>> http://manning.com/ghosh2
>>> http://manning.com/ghosh
>>>
>>> Twttr: @debasishg
>>> Blog: http://debasishg.blogspot.com
>>> Code: http://github.com/debasishg
>>>
>>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: Kafka Streams state store issue on cluster
Posted by Damian Guy <da...@gmail.com>.
Do you have any logs that might help to work out what is going wrong?
On Fri, 28 Jul 2017 at 14:16 Damian Guy <da...@gmail.com> wrote:
> The config looks ok to me
>
> On Fri, 28 Jul 2017 at 13:24 Debasish Ghosh <gh...@gmail.com>
> wrote:
>
>> I am setting APPLICATION_SERVER_CONFIG, which is possibly what u r
>> referring to. Just now I noticed that I may also need to set
>> REPLICATION_FACTOR_CONFIG, which needs to be set to 2 (default is 1).
>> Anything else that I may be missing ?
>>
>>
>> regards.
>>
>> On Fri, Jul 28, 2017 at 5:46 PM, Debasish Ghosh <ghosh.debasish@gmail.com
>> >
>> wrote:
>>
>> > Hi Damien -
>> >
>> > I am not sure I understand what u mean .. I have the following set in
>> the
>> > application .. Do I need to set anything else at the host level ?
>> > Environment variable ?
>> >
>> > val streamingConfig = {
>> > val settings = new Properties
>> > settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
>> > "kstream-weblog-processing")
>> > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
>> config.brokers)
>> >
>> > config.schemaRegistryUrl.foreach{ url =>
>> >
>> settings.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
>> > url)
>> > }
>> >
>> > settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
>> > Serdes.ByteArray.getClass.getName)
>> > settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
>> > Serdes.String.getClass.getName)
>> >
>> > // setting offset reset to earliest so that we can re-run the demo
>> > code with the same pre-loaded data
>> > // Note: To re-run the demo, you need to use the offset reset
>> tool:
>> > // https://cwiki.apache.org/confluence/display/KAFKA/
>> > Kafka+Streams+Application+Reset+Tool
>> > settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
>> >
>> > // need this for query service
>> > settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
>> > s"${config.httpInterface}:${config.httpPort}")
>> >
>> > // default is /tmp/kafka-streams
>> > settings.put(StreamsConfig.STATE_DIR_CONFIG, config.stateStoreDir)
>> >
>> > // Set the commit interval to 500ms so that any changes are
>> flushed
>> > frequently and the summary
>> > // data are updated with low latency.
>> > settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
>> >
>> > settings
>> > }
>> >
>> > Please explain a bit ..
>> >
>> > regards.
>> >
>> >
>> > On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com>
>> wrote:
>> >
>> >> Hi,
>> >>
>> >> Do you have the application.server property set appropriately for both
>> >> hosts?
>> >>
>> >> The second stack trace is this bug:
>> >> https://issues.apache.org/jira/browse/KAFKA-5556
>> >>
>> >> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <gh...@gmail.com>
>> >> wrote:
>> >>
>> >> > Hi -
>> >> >
>> >> > In my Kafka Streams application, I have a state store resulting from
>> a
>> >> > stateful streaming topology. The environment is
>> >> >
>> >> > - Kafka 0.10.2.1
>> >> > - It runs on a DC/OS cluster
>> >> > - I am running Confluent-Kafka 3.2.2 on the cluster
>> >> > - Each topic that I have has 2 partitions with replication factor
>> = 2
>> >> > - The application also has an associated http service that does
>> >> > interactive queries on the state store
>> >> >
>> >> > The application runs fine when I invoke a single instance. I can use
>> the
>> >> > http endpoints to do queries and everything looks good. Problems
>> surface
>> >> > when I try to spawn another instance of the application. I use the
>> same
>> >> > APPLICATION_ID_CONFIG and the instance starts on a different node of
>> the
>> >> > cluster. The data consumption part works fine as the new instance
>> also
>> >> > starts consuming from the same topic as the first one. But when I try
>> >> the
>> >> > http query, the metadata fetch fails ..
>> >> >
>> >> > I have some code snippet like this as part of the query that tries to
>> >> fetch
>> >> > the metadata so that I can locate the host to query on ..
>> >> >
>> >> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
>> >> > stringSerializer) match {
>> >> > case Success(host) => {
>> >> > // hostKey is on another instance. call the other instance to
>> >> fetch
>> >> > the data.
>> >> > if (!thisHost(host)) {
>> >> > logger.warn(s"Key $hostKey is on another instance not on
>> >> $host -
>> >> > requerying ..")
>> >> > httpRequester.queryFromHost[Long](host, path)
>> >> > } else {
>> >> > // hostKey is on this instance
>> >> > localStateStoreQuery.queryStateStore(streams, store,
>> hostKey)
>> >> > }
>> >> > }
>> >> > case Failure(ex) => Future.failed(ex)
>> >> > }
>> >> >
>> >> > and the metadataService.streamsMetadataForStoreAndKey has the
>> following
>> >> > call ..
>> >> >
>> >> > streams.metadataForKey(store, key, serializer) match {
>> >> > case null => throw new IllegalArgumentException(s"Metadata for
>> >> key
>> >> > $key not found in $store")
>> >> > case metadata => new HostStoreInfo(metadata.host,
>> metadata.port,
>> >> > metadata.stateStoreNames.asScala.toSet)
>> >> > }
>> >> >
>> >> > When I start the second instance, streams.metadataForKey returns null
>> >> for
>> >> > any key I pass .. Here's the relevant stack trace ..
>> >> >
>> >> > java.lang.IllegalArgumentException: Metadata for key mtc.clark.net
>> not
>> >> > found in access-count-per-host
>> >> > at
>> >> >
>> >> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
>> >> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
>> >> > at scala.util.Try$.apply(Try.scala:209)
>> >> > at
>> >> >
>> >> > com.xx.fdp.sample.kstream.services.MetadataService.streamsMe
>> >> tadataForStoreAndKey(MetadataService.scala:46)
>> >> > at
>> >> >
>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryI
>> >> nfo(KeyValueFetcher.scala:36)
>> >> > at
>> >> >
>> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCo
>> >> untSummary(KeyValueFetcher.scala:29)
>> >> > at
>> >> >
>> >> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun
>> >> $routes$10(WeblogDSLHttpService.scala:41)
>> >> > at
>> >> >
>> >> > akka.http.scaladsl.server.directives.RouteDirectives.$anonfu
>> >> n$complete$1(RouteDirectives.scala:47)
>> >> > at
>> >> >
>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>> >> rdRoute.scala:19)
>> >> > at
>> >> >
>> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>> >> rdRoute.scala:19)
>> >> > ...
>> >> >
>> >> > and following this exception I get another one which looks like an
>> >> internal
>> >> > exception that stops the application ..
>> >> >
>> >> > 09:57:33.731 TKD [StreamThread-1] ERROR
>> o.a.k.s.p.internals.StreamThread
>> >> -
>> >> > stream-thread [StreamThread-1] Streams application error during
>> >> processing:
>> >> > java.lang.IllegalStateException: Attempt to retrieve exception from
>> >> future
>> >> > which hasn't failed
>> >> > at
>> >> >
>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>> >> exception(RequestFuture.java:99)
>> >> > at
>> >> >
>> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>> >> isRetriable(RequestFuture.java:89)
>> >> > at
>> >> >
>> >> > org.apache.kafka.clients.consumer.internals.ConsumerCoordina
>> >> tor.commitOffsetsSync(ConsumerCoordinator.java:590)
>> >> > at
>> >> >
>> >> > org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(
>> >> KafkaConsumer.java:1124)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>> >> commitOffsets(StreamTask.java:296)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamTask$1.
>> >> run(StreamTask.java:79)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamsMetricsI
>> >> mpl.measureLatencyNs(StreamsMetricsImpl.java:188)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamTask.
>> >> commit(StreamTask.java:280)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>> >> commitOne(StreamThread.java:807)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>> >> commitAll(StreamThread.java:794)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>> >> maybeCommit(StreamThread.java:769)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>> >> runLoop(StreamThread.java:647)
>> >> > at
>> >> >
>> >> > org.apache.kafka.streams.processor.internals.StreamThread.
>> >> run(StreamThread.java:361)
>> >> >
>> >> > BTW the application runs fine when I have 2 instances running on the
>> >> same
>> >> > host (my laptop) or on 1 instance on the cluster. The problem
>> surfaces
>> >> only
>> >> > when I start the second instance.
>> >> >
>> >> > Any help / pointer / area to look into ?
>> >> >
>> >> > --
>> >> > Debasish Ghosh
>> >> > http://manning.com/ghosh2
>> >> > http://manning.com/ghosh
>> >> >
>> >> > Twttr: @debasishg
>> >> > Blog: http://debasishg.blogspot.com
>> >> > Code: http://github.com/debasishg
>> >> >
>> >>
>> >
>> >
>> >
>> > --
>> > Debasish Ghosh
>> > http://manning.com/ghosh2
>> > http://manning.com/ghosh
>> >
>> > Twttr: @debasishg
>> > Blog: http://debasishg.blogspot.com
>> > Code: http://github.com/debasishg
>> >
>>
>>
>>
>> --
>> Debasish Ghosh
>> http://manning.com/ghosh2
>> http://manning.com/ghosh
>>
>> Twttr: @debasishg
>> Blog: http://debasishg.blogspot.com
>> Code: http://github.com/debasishg
>>
>
Re: Kafka Streams state store issue on cluster
Posted by Damian Guy <da...@gmail.com>.
The config looks ok to me
On Fri, 28 Jul 2017 at 13:24 Debasish Ghosh <gh...@gmail.com>
wrote:
> I am setting APPLICATION_SERVER_CONFIG, which is possibly what u r
> referring to. Just now I noticed that I may also need to set
> REPLICATION_FACTOR_CONFIG, which needs to be set to 2 (default is 1).
> Anything else that I may be missing ?
>
>
> regards.
>
> On Fri, Jul 28, 2017 at 5:46 PM, Debasish Ghosh <gh...@gmail.com>
> wrote:
>
> > Hi Damien -
> >
> > I am not sure I understand what u mean .. I have the following set in the
> > application .. Do I need to set anything else at the host level ?
> > Environment variable ?
> >
> > val streamingConfig = {
> > val settings = new Properties
> > settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
> > "kstream-weblog-processing")
> > settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG,
> config.brokers)
> >
> > config.schemaRegistryUrl.foreach{ url =>
> >
> settings.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
> > url)
> > }
> >
> > settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
> > Serdes.ByteArray.getClass.getName)
> > settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
> > Serdes.String.getClass.getName)
> >
> > // setting offset reset to earliest so that we can re-run the demo
> > code with the same pre-loaded data
> > // Note: To re-run the demo, you need to use the offset reset tool:
> > // https://cwiki.apache.org/confluence/display/KAFKA/
> > Kafka+Streams+Application+Reset+Tool
> > settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
> >
> > // need this for query service
> > settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
> > s"${config.httpInterface}:${config.httpPort}")
> >
> > // default is /tmp/kafka-streams
> > settings.put(StreamsConfig.STATE_DIR_CONFIG, config.stateStoreDir)
> >
> > // Set the commit interval to 500ms so that any changes are flushed
> > frequently and the summary
> > // data are updated with low latency.
> > settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
> >
> > settings
> > }
> >
> > Please explain a bit ..
> >
> > regards.
> >
> >
> > On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com>
> wrote:
> >
> >> Hi,
> >>
> >> Do you have the application.server property set appropriately for both
> >> hosts?
> >>
> >> The second stack trace is this bug:
> >> https://issues.apache.org/jira/browse/KAFKA-5556
> >>
> >> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <gh...@gmail.com>
> >> wrote:
> >>
> >> > Hi -
> >> >
> >> > In my Kafka Streams application, I have a state store resulting from a
> >> > stateful streaming topology. The environment is
> >> >
> >> > - Kafka 0.10.2.1
> >> > - It runs on a DC/OS cluster
> >> > - I am running Confluent-Kafka 3.2.2 on the cluster
> >> > - Each topic that I have has 2 partitions with replication factor
> = 2
> >> > - The application also has an associated http service that does
> >> > interactive queries on the state store
> >> >
> >> > The application runs fine when I invoke a single instance. I can use
> the
> >> > http endpoints to do queries and everything looks good. Problems
> surface
> >> > when I try to spawn another instance of the application. I use the
> same
> >> > APPLICATION_ID_CONFIG and the instance starts on a different node of
> the
> >> > cluster. The data consumption part works fine as the new instance also
> >> > starts consuming from the same topic as the first one. But when I try
> >> the
> >> > http query, the metadata fetch fails ..
> >> >
> >> > I have some code snippet like this as part of the query that tries to
> >> fetch
> >> > the metadata so that I can locate the host to query on ..
> >> >
> >> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
> >> > stringSerializer) match {
> >> > case Success(host) => {
> >> > // hostKey is on another instance. call the other instance to
> >> fetch
> >> > the data.
> >> > if (!thisHost(host)) {
> >> > logger.warn(s"Key $hostKey is on another instance not on
> >> $host -
> >> > requerying ..")
> >> > httpRequester.queryFromHost[Long](host, path)
> >> > } else {
> >> > // hostKey is on this instance
> >> > localStateStoreQuery.queryStateStore(streams, store,
> hostKey)
> >> > }
> >> > }
> >> > case Failure(ex) => Future.failed(ex)
> >> > }
> >> >
> >> > and the metadataService.streamsMetadataForStoreAndKey has the
> following
> >> > call ..
> >> >
> >> > streams.metadataForKey(store, key, serializer) match {
> >> > case null => throw new IllegalArgumentException(s"Metadata for
> >> key
> >> > $key not found in $store")
> >> > case metadata => new HostStoreInfo(metadata.host, metadata.port,
> >> > metadata.stateStoreNames.asScala.toSet)
> >> > }
> >> >
> >> > When I start the second instance, streams.metadataForKey returns null
> >> for
> >> > any key I pass .. Here's the relevant stack trace ..
> >> >
> >> > java.lang.IllegalArgumentException: Metadata for key mtc.clark.net
> not
> >> > found in access-count-per-host
> >> > at
> >> >
> >> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
> >> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
> >> > at scala.util.Try$.apply(Try.scala:209)
> >> > at
> >> >
> >> > com.xx.fdp.sample.kstream.services.MetadataService.streamsMe
> >> tadataForStoreAndKey(MetadataService.scala:46)
> >> > at
> >> >
> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryI
> >> nfo(KeyValueFetcher.scala:36)
> >> > at
> >> >
> >> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCo
> >> untSummary(KeyValueFetcher.scala:29)
> >> > at
> >> >
> >> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun
> >> $routes$10(WeblogDSLHttpService.scala:41)
> >> > at
> >> >
> >> > akka.http.scaladsl.server.directives.RouteDirectives.$anonfu
> >> n$complete$1(RouteDirectives.scala:47)
> >> > at
> >> >
> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
> >> rdRoute.scala:19)
> >> > at
> >> >
> >> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
> >> rdRoute.scala:19)
> >> > ...
> >> >
> >> > and following this exception I get another one which looks like an
> >> internal
> >> > exception that stops the application ..
> >> >
> >> > 09:57:33.731 TKD [StreamThread-1] ERROR
> o.a.k.s.p.internals.StreamThread
> >> -
> >> > stream-thread [StreamThread-1] Streams application error during
> >> processing:
> >> > java.lang.IllegalStateException: Attempt to retrieve exception from
> >> future
> >> > which hasn't failed
> >> > at
> >> >
> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
> >> exception(RequestFuture.java:99)
> >> > at
> >> >
> >> > org.apache.kafka.clients.consumer.internals.RequestFuture.
> >> isRetriable(RequestFuture.java:89)
> >> > at
> >> >
> >> > org.apache.kafka.clients.consumer.internals.ConsumerCoordina
> >> tor.commitOffsetsSync(ConsumerCoordinator.java:590)
> >> > at
> >> >
> >> > org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(
> >> KafkaConsumer.java:1124)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamTask.
> >> commitOffsets(StreamTask.java:296)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamTask$1.
> >> run(StreamTask.java:79)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamsMetricsI
> >> mpl.measureLatencyNs(StreamsMetricsImpl.java:188)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamTask.
> >> commit(StreamTask.java:280)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamThread.
> >> commitOne(StreamThread.java:807)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamThread.
> >> commitAll(StreamThread.java:794)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamThread.
> >> maybeCommit(StreamThread.java:769)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamThread.
> >> runLoop(StreamThread.java:647)
> >> > at
> >> >
> >> > org.apache.kafka.streams.processor.internals.StreamThread.
> >> run(StreamThread.java:361)
> >> >
> >> > BTW the application runs fine when I have 2 instances running on the
> >> same
> >> > host (my laptop) or on 1 instance on the cluster. The problem surfaces
> >> only
> >> > when I start the second instance.
> >> >
> >> > Any help / pointer / area to look into ?
> >> >
> >> > --
> >> > Debasish Ghosh
> >> > http://manning.com/ghosh2
> >> > http://manning.com/ghosh
> >> >
> >> > Twttr: @debasishg
> >> > Blog: http://debasishg.blogspot.com
> >> > Code: http://github.com/debasishg
> >> >
> >>
> >
> >
> >
> > --
> > Debasish Ghosh
> > http://manning.com/ghosh2
> > http://manning.com/ghosh
> >
> > Twttr: @debasishg
> > Blog: http://debasishg.blogspot.com
> > Code: http://github.com/debasishg
> >
>
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>
Re: Kafka Streams state store issue on cluster
Posted by Debasish Ghosh <gh...@gmail.com>.
I am setting APPLICATION_SERVER_CONFIG, which is possibly what u r
referring to. Just now I noticed that I may also need to set
REPLICATION_FACTOR_CONFIG, which needs to be set to 2 (default is 1).
Anything else that I may be missing ?
regards.
On Fri, Jul 28, 2017 at 5:46 PM, Debasish Ghosh <gh...@gmail.com>
wrote:
> Hi Damien -
>
> I am not sure I understand what u mean .. I have the following set in the
> application .. Do I need to set anything else at the host level ?
> Environment variable ?
>
> val streamingConfig = {
> val settings = new Properties
> settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
> "kstream-weblog-processing")
> settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokers)
>
> config.schemaRegistryUrl.foreach{ url =>
> settings.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG,
> url)
> }
>
> settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
> Serdes.ByteArray.getClass.getName)
> settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
> Serdes.String.getClass.getName)
>
> // setting offset reset to earliest so that we can re-run the demo
> code with the same pre-loaded data
> // Note: To re-run the demo, you need to use the offset reset tool:
> // https://cwiki.apache.org/confluence/display/KAFKA/
> Kafka+Streams+Application+Reset+Tool
> settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
>
> // need this for query service
> settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
> s"${config.httpInterface}:${config.httpPort}")
>
> // default is /tmp/kafka-streams
> settings.put(StreamsConfig.STATE_DIR_CONFIG, config.stateStoreDir)
>
> // Set the commit interval to 500ms so that any changes are flushed
> frequently and the summary
> // data are updated with low latency.
> settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
>
> settings
> }
>
> Please explain a bit ..
>
> regards.
>
>
> On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com> wrote:
>
>> Hi,
>>
>> Do you have the application.server property set appropriately for both
>> hosts?
>>
>> The second stack trace is this bug:
>> https://issues.apache.org/jira/browse/KAFKA-5556
>>
>> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <gh...@gmail.com>
>> wrote:
>>
>> > Hi -
>> >
>> > In my Kafka Streams application, I have a state store resulting from a
>> > stateful streaming topology. The environment is
>> >
>> > - Kafka 0.10.2.1
>> > - It runs on a DC/OS cluster
>> > - I am running Confluent-Kafka 3.2.2 on the cluster
>> > - Each topic that I have has 2 partitions with replication factor = 2
>> > - The application also has an associated http service that does
>> > interactive queries on the state store
>> >
>> > The application runs fine when I invoke a single instance. I can use the
>> > http endpoints to do queries and everything looks good. Problems surface
>> > when I try to spawn another instance of the application. I use the same
>> > APPLICATION_ID_CONFIG and the instance starts on a different node of the
>> > cluster. The data consumption part works fine as the new instance also
>> > starts consuming from the same topic as the first one. But when I try
>> the
>> > http query, the metadata fetch fails ..
>> >
>> > I have some code snippet like this as part of the query that tries to
>> fetch
>> > the metadata so that I can locate the host to query on ..
>> >
>> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
>> > stringSerializer) match {
>> > case Success(host) => {
>> > // hostKey is on another instance. call the other instance to
>> fetch
>> > the data.
>> > if (!thisHost(host)) {
>> > logger.warn(s"Key $hostKey is on another instance not on
>> $host -
>> > requerying ..")
>> > httpRequester.queryFromHost[Long](host, path)
>> > } else {
>> > // hostKey is on this instance
>> > localStateStoreQuery.queryStateStore(streams, store, hostKey)
>> > }
>> > }
>> > case Failure(ex) => Future.failed(ex)
>> > }
>> >
>> > and the metadataService.streamsMetadataForStoreAndKey has the following
>> > call ..
>> >
>> > streams.metadataForKey(store, key, serializer) match {
>> > case null => throw new IllegalArgumentException(s"Metadata for
>> key
>> > $key not found in $store")
>> > case metadata => new HostStoreInfo(metadata.host, metadata.port,
>> > metadata.stateStoreNames.asScala.toSet)
>> > }
>> >
>> > When I start the second instance, streams.metadataForKey returns null
>> for
>> > any key I pass .. Here's the relevant stack trace ..
>> >
>> > java.lang.IllegalArgumentException: Metadata for key mtc.clark.net not
>> > found in access-count-per-host
>> > at
>> >
>> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
>> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
>> > at scala.util.Try$.apply(Try.scala:209)
>> > at
>> >
>> > com.xx.fdp.sample.kstream.services.MetadataService.streamsMe
>> tadataForStoreAndKey(MetadataService.scala:46)
>> > at
>> >
>> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryI
>> nfo(KeyValueFetcher.scala:36)
>> > at
>> >
>> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCo
>> untSummary(KeyValueFetcher.scala:29)
>> > at
>> >
>> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun
>> $routes$10(WeblogDSLHttpService.scala:41)
>> > at
>> >
>> > akka.http.scaladsl.server.directives.RouteDirectives.$anonfu
>> n$complete$1(RouteDirectives.scala:47)
>> > at
>> >
>> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>> rdRoute.scala:19)
>> > at
>> >
>> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(Standa
>> rdRoute.scala:19)
>> > ...
>> >
>> > and following this exception I get another one which looks like an
>> internal
>> > exception that stops the application ..
>> >
>> > 09:57:33.731 TKD [StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread
>> -
>> > stream-thread [StreamThread-1] Streams application error during
>> processing:
>> > java.lang.IllegalStateException: Attempt to retrieve exception from
>> future
>> > which hasn't failed
>> > at
>> >
>> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>> exception(RequestFuture.java:99)
>> > at
>> >
>> > org.apache.kafka.clients.consumer.internals.RequestFuture.
>> isRetriable(RequestFuture.java:89)
>> > at
>> >
>> > org.apache.kafka.clients.consumer.internals.ConsumerCoordina
>> tor.commitOffsetsSync(ConsumerCoordinator.java:590)
>> > at
>> >
>> > org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(
>> KafkaConsumer.java:1124)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamTask.
>> commitOffsets(StreamTask.java:296)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamTask$1.
>> run(StreamTask.java:79)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamsMetricsI
>> mpl.measureLatencyNs(StreamsMetricsImpl.java:188)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamTask.
>> commit(StreamTask.java:280)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamThread.
>> commitOne(StreamThread.java:807)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamThread.
>> commitAll(StreamThread.java:794)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamThread.
>> maybeCommit(StreamThread.java:769)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamThread.
>> runLoop(StreamThread.java:647)
>> > at
>> >
>> > org.apache.kafka.streams.processor.internals.StreamThread.
>> run(StreamThread.java:361)
>> >
>> > BTW the application runs fine when I have 2 instances running on the
>> same
>> > host (my laptop) or on 1 instance on the cluster. The problem surfaces
>> only
>> > when I start the second instance.
>> >
>> > Any help / pointer / area to look into ?
>> >
>> > --
>> > Debasish Ghosh
>> > http://manning.com/ghosh2
>> > http://manning.com/ghosh
>> >
>> > Twttr: @debasishg
>> > Blog: http://debasishg.blogspot.com
>> > Code: http://github.com/debasishg
>> >
>>
>
>
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: Kafka Streams state store issue on cluster
Posted by Debasish Ghosh <gh...@gmail.com>.
Hi Damien -
I am not sure I understand what u mean .. I have the following set in the
application .. Do I need to set anything else at the host level ?
Environment variable ?
val streamingConfig = {
val settings = new Properties
settings.put(StreamsConfig.APPLICATION_ID_CONFIG,
"kstream-weblog-processing")
settings.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, config.brokers)
config.schemaRegistryUrl.foreach{ url =>
settings.put(AbstractKafkaAvroSerDeConfig.SCHEMA_REGISTRY_URL_CONFIG, url)
}
settings.put(StreamsConfig.KEY_SERDE_CLASS_CONFIG,
Serdes.ByteArray.getClass.getName)
settings.put(StreamsConfig.VALUE_SERDE_CLASS_CONFIG,
Serdes.String.getClass.getName)
// setting offset reset to earliest so that we can re-run the demo
code with the same pre-loaded data
// Note: To re-run the demo, you need to use the offset reset tool:
//
https://cwiki.apache.org/confluence/display/KAFKA/Kafka+Streams+Application+Reset+Tool
settings.put(ConsumerConfig.AUTO_OFFSET_RESET_CONFIG, "earliest")
// need this for query service
settings.put(StreamsConfig.APPLICATION_SERVER_CONFIG,
s"${config.httpInterface}:${config.httpPort}")
// default is /tmp/kafka-streams
settings.put(StreamsConfig.STATE_DIR_CONFIG, config.stateStoreDir)
// Set the commit interval to 500ms so that any changes are flushed
frequently and the summary
// data are updated with low latency.
settings.put(StreamsConfig.COMMIT_INTERVAL_MS_CONFIG, "500")
settings
}
Please explain a bit ..
regards.
On Fri, Jul 28, 2017 at 5:36 PM, Damian Guy <da...@gmail.com> wrote:
> Hi,
>
> Do you have the application.server property set appropriately for both
> hosts?
>
> The second stack trace is this bug:
> https://issues.apache.org/jira/browse/KAFKA-5556
>
> On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <gh...@gmail.com>
> wrote:
>
> > Hi -
> >
> > In my Kafka Streams application, I have a state store resulting from a
> > stateful streaming topology. The environment is
> >
> > - Kafka 0.10.2.1
> > - It runs on a DC/OS cluster
> > - I am running Confluent-Kafka 3.2.2 on the cluster
> > - Each topic that I have has 2 partitions with replication factor = 2
> > - The application also has an associated http service that does
> > interactive queries on the state store
> >
> > The application runs fine when I invoke a single instance. I can use the
> > http endpoints to do queries and everything looks good. Problems surface
> > when I try to spawn another instance of the application. I use the same
> > APPLICATION_ID_CONFIG and the instance starts on a different node of the
> > cluster. The data consumption part works fine as the new instance also
> > starts consuming from the same topic as the first one. But when I try the
> > http query, the metadata fetch fails ..
> >
> > I have some code snippet like this as part of the query that tries to
> fetch
> > the metadata so that I can locate the host to query on ..
> >
> > metadataService.streamsMetadataForStoreAndKey(store, hostKey,
> > stringSerializer) match {
> > case Success(host) => {
> > // hostKey is on another instance. call the other instance to
> fetch
> > the data.
> > if (!thisHost(host)) {
> > logger.warn(s"Key $hostKey is on another instance not on $host
> -
> > requerying ..")
> > httpRequester.queryFromHost[Long](host, path)
> > } else {
> > // hostKey is on this instance
> > localStateStoreQuery.queryStateStore(streams, store, hostKey)
> > }
> > }
> > case Failure(ex) => Future.failed(ex)
> > }
> >
> > and the metadataService.streamsMetadataForStoreAndKey has the following
> > call ..
> >
> > streams.metadataForKey(store, key, serializer) match {
> > case null => throw new IllegalArgumentException(s"Metadata for key
> > $key not found in $store")
> > case metadata => new HostStoreInfo(metadata.host, metadata.port,
> > metadata.stateStoreNames.asScala.toSet)
> > }
> >
> > When I start the second instance, streams.metadataForKey returns null for
> > any key I pass .. Here's the relevant stack trace ..
> >
> > java.lang.IllegalArgumentException: Metadata for key mtc.clark.net not
> > found in access-count-per-host
> > at
> >
> > com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$
> streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
> > at scala.util.Try$.apply(Try.scala:209)
> > at
> >
> > com.xx.fdp.sample.kstream.services.MetadataService.
> streamsMetadataForStoreAndKey(MetadataService.scala:46)
> > at
> >
> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryInfo(
> KeyValueFetcher.scala:36)
> > at
> >
> > com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCountSummary(
> KeyValueFetcher.scala:29)
> > at
> >
> > com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun$routes$10(
> WeblogDSLHttpService.scala:41)
> > at
> >
> > akka.http.scaladsl.server.directives.RouteDirectives.$
> anonfun$complete$1(RouteDirectives.scala:47)
> > at
> >
> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(
> StandardRoute.scala:19)
> > at
> >
> > akka.http.scaladsl.server.StandardRoute$$anon$1.apply(
> StandardRoute.scala:19)
> > ...
> >
> > and following this exception I get another one which looks like an
> internal
> > exception that stops the application ..
> >
> > 09:57:33.731 TKD [StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread
> -
> > stream-thread [StreamThread-1] Streams application error during
> processing:
> > java.lang.IllegalStateException: Attempt to retrieve exception from
> future
> > which hasn't failed
> > at
> >
> > org.apache.kafka.clients.consumer.internals.RequestFuture.exception(
> RequestFuture.java:99)
> > at
> >
> > org.apache.kafka.clients.consumer.internals.RequestFuture.isRetriable(
> RequestFuture.java:89)
> > at
> >
> > org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.
> commitOffsetsSync(ConsumerCoordinator.java:590)
> > at
> >
> > org.apache.kafka.clients.consumer.KafkaConsumer.
> commitSync(KafkaConsumer.java:1124)
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(
> StreamTask.java:296)
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamTask$1.run(StreamTask.java:79)
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.
> measureLatencyNs(StreamsMetricsImpl.java:188)
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamTask.commit(StreamTask.java:280)
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamThread.commitOne(
> StreamThread.java:807)
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamThread.commitAll(
> StreamThread.java:794)
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(
> StreamThread.java:769)
> > at
> >
> > org.apache.kafka.streams.processor.internals.StreamThread.runLoop(
> StreamThread.java:647)
> > at
> >
> > org.apache.kafka.streams.processor.internals.
> StreamThread.run(StreamThread.java:361)
> >
> > BTW the application runs fine when I have 2 instances running on the same
> > host (my laptop) or on 1 instance on the cluster. The problem surfaces
> only
> > when I start the second instance.
> >
> > Any help / pointer / area to look into ?
> >
> > --
> > Debasish Ghosh
> > http://manning.com/ghosh2
> > http://manning.com/ghosh
> >
> > Twttr: @debasishg
> > Blog: http://debasishg.blogspot.com
> > Code: http://github.com/debasishg
> >
>
--
Debasish Ghosh
http://manning.com/ghosh2
http://manning.com/ghosh
Twttr: @debasishg
Blog: http://debasishg.blogspot.com
Code: http://github.com/debasishg
Re: Kafka Streams state store issue on cluster
Posted by Damian Guy <da...@gmail.com>.
Hi,
Do you have the application.server property set appropriately for both
hosts?
The second stack trace is this bug:
https://issues.apache.org/jira/browse/KAFKA-5556
On Fri, 28 Jul 2017 at 12:55 Debasish Ghosh <gh...@gmail.com>
wrote:
> Hi -
>
> In my Kafka Streams application, I have a state store resulting from a
> stateful streaming topology. The environment is
>
> - Kafka 0.10.2.1
> - It runs on a DC/OS cluster
> - I am running Confluent-Kafka 3.2.2 on the cluster
> - Each topic that I have has 2 partitions with replication factor = 2
> - The application also has an associated http service that does
> interactive queries on the state store
>
> The application runs fine when I invoke a single instance. I can use the
> http endpoints to do queries and everything looks good. Problems surface
> when I try to spawn another instance of the application. I use the same
> APPLICATION_ID_CONFIG and the instance starts on a different node of the
> cluster. The data consumption part works fine as the new instance also
> starts consuming from the same topic as the first one. But when I try the
> http query, the metadata fetch fails ..
>
> I have some code snippet like this as part of the query that tries to fetch
> the metadata so that I can locate the host to query on ..
>
> metadataService.streamsMetadataForStoreAndKey(store, hostKey,
> stringSerializer) match {
> case Success(host) => {
> // hostKey is on another instance. call the other instance to fetch
> the data.
> if (!thisHost(host)) {
> logger.warn(s"Key $hostKey is on another instance not on $host -
> requerying ..")
> httpRequester.queryFromHost[Long](host, path)
> } else {
> // hostKey is on this instance
> localStateStoreQuery.queryStateStore(streams, store, hostKey)
> }
> }
> case Failure(ex) => Future.failed(ex)
> }
>
> and the metadataService.streamsMetadataForStoreAndKey has the following
> call ..
>
> streams.metadataForKey(store, key, serializer) match {
> case null => throw new IllegalArgumentException(s"Metadata for key
> $key not found in $store")
> case metadata => new HostStoreInfo(metadata.host, metadata.port,
> metadata.stateStoreNames.asScala.toSet)
> }
>
> When I start the second instance, streams.metadataForKey returns null for
> any key I pass .. Here's the relevant stack trace ..
>
> java.lang.IllegalArgumentException: Metadata for key mtc.clark.net not
> found in access-count-per-host
> at
>
> com.xx.fdp.sample.kstream.services.MetadataService.$anonfun$streamsMetadataForStoreAndKey$1(MetadataService.scala:51)
> at scala.util.Try$.apply(Try.scala:209)
> at
>
> com.xx.fdp.sample.kstream.services.MetadataService.streamsMetadataForStoreAndKey(MetadataService.scala:46)
> at
>
> com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchSummaryInfo(KeyValueFetcher.scala:36)
> at
>
> com.xx.fdp.sample.kstream.http.KeyValueFetcher.fetchAccessCountSummary(KeyValueFetcher.scala:29)
> at
>
> com.xx.fdp.sample.kstream.http.WeblogDSLHttpService.$anonfun$routes$10(WeblogDSLHttpService.scala:41)
> at
>
> akka.http.scaladsl.server.directives.RouteDirectives.$anonfun$complete$1(RouteDirectives.scala:47)
> at
>
> akka.http.scaladsl.server.StandardRoute$$anon$1.apply(StandardRoute.scala:19)
> at
>
> akka.http.scaladsl.server.StandardRoute$$anon$1.apply(StandardRoute.scala:19)
> ...
>
> and following this exception I get another one which looks like an internal
> exception that stops the application ..
>
> 09:57:33.731 TKD [StreamThread-1] ERROR o.a.k.s.p.internals.StreamThread -
> stream-thread [StreamThread-1] Streams application error during processing:
> java.lang.IllegalStateException: Attempt to retrieve exception from future
> which hasn't failed
> at
>
> org.apache.kafka.clients.consumer.internals.RequestFuture.exception(RequestFuture.java:99)
> at
>
> org.apache.kafka.clients.consumer.internals.RequestFuture.isRetriable(RequestFuture.java:89)
> at
>
> org.apache.kafka.clients.consumer.internals.ConsumerCoordinator.commitOffsetsSync(ConsumerCoordinator.java:590)
> at
>
> org.apache.kafka.clients.consumer.KafkaConsumer.commitSync(KafkaConsumer.java:1124)
> at
>
> org.apache.kafka.streams.processor.internals.StreamTask.commitOffsets(StreamTask.java:296)
> at
>
> org.apache.kafka.streams.processor.internals.StreamTask$1.run(StreamTask.java:79)
> at
>
> org.apache.kafka.streams.processor.internals.StreamsMetricsImpl.measureLatencyNs(StreamsMetricsImpl.java:188)
> at
>
> org.apache.kafka.streams.processor.internals.StreamTask.commit(StreamTask.java:280)
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.commitOne(StreamThread.java:807)
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.commitAll(StreamThread.java:794)
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.maybeCommit(StreamThread.java:769)
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.runLoop(StreamThread.java:647)
> at
>
> org.apache.kafka.streams.processor.internals.StreamThread.run(StreamThread.java:361)
>
> BTW the application runs fine when I have 2 instances running on the same
> host (my laptop) or on 1 instance on the cluster. The problem surfaces only
> when I start the second instance.
>
> Any help / pointer / area to look into ?
>
> --
> Debasish Ghosh
> http://manning.com/ghosh2
> http://manning.com/ghosh
>
> Twttr: @debasishg
> Blog: http://debasishg.blogspot.com
> Code: http://github.com/debasishg
>