The config looks ok to me
On Fri, 28 Jul 2017 at 13:24 Debasish Ghosh <ghosh.debas...@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.debas...@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 <damian....@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.debas...@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
>
