Hey Jacob, right you are.
I added a "*<logger name="org.apache.samza.container" level="DEBUG"/>*"
logback configuration. And lo and behold, this exception detail popped up.
10:05:47.520 [main] DEBUG o.a.samza.container.SamzaContainer$ - Exception
detail:
java.lang.RuntimeException: Could not instantiate class
*org.apache.samza.system.elasticsearch.clientTransport.ClientFactory*
at
org.apache.samza.system.elasticsearch.ElasticsearchSystemFactory.getClient(ElasticsearchSystemFactory.java:79)
at
org.apache.samza.system.elasticsearch.ElasticsearchSystemFactory.getProducer(ElasticsearchSystemFactory.java:54)
at
org.apache.samza.container.SamzaContainer$$anonfun$13.apply(SamzaContainer.scala:195)
...
at clojure.lang.Var.invoke(Var.java:383)
at clojure.lang.AFn.applyToHelper(AFn.java:156)
at clojure.lang.Var.applyTo(Var.java:700)
at clojure.main.main(main.java:37)
I fixed the typo (in bold red), and that solved the problem :)
Cheers
Tim
On 10 May 2016 at 09:08, Jacob Maes <jacob.m...@gmail.com> wrote:
> Hey Tim,
>
> Can you include the full log? Since the producer is not showing up, I'd
> expect to see some earlier messages explaining why. In particular, this
> section of code from SamzaContainer.scala should print errors instantiating
> the producers:
>
> val producers = systemFactories
> > .map {
> > case (systemName, systemFactory) =>
> > try {
> > (systemName, systemFactory.getProducer(systemName, config,
> > samzaContainerMetrics.registry))
> > } catch {
> > case e: Exception =>
> > info("Failed to create a producer for %s, so skipping."
> > format systemName)
> > debug("Exception detail:", e)
> > (systemName, null)
> > }
> > }
> > .filter(_._2 != null)
> > .toMap
> > info("Got system producers: %s" format producers.keys)
>
>
> And from the looks of it, it would be a good idea to enable debug logging
> at least for org.apache.samza.container, so we can see the exception
> detail.
>
> Thanks,
> Jake
>
> On Tue, May 10, 2016 at 8:36 AM, Tim Washington <
> tim.washing...@fundingcircle.com> wrote:
>
> > *Tim Washington* | *Senior Software Engineer*
> >
> > *tim.washing...@fundingcircle.com <renee.nich...@fundingcircle.com>*
> >
> > 747 Front St, 4th Fl | San Francisco, CA 94111
> >
> > *Our Mission: **T**o build a better financial world*
> >
> >
> >
> >
> > ---------- Forwarded message ----------
> > From: Tim Washington <tim.washing...@fundingcircle.com>
> > Date: 9 May 2016 at 15:40
> > Subject: Howto Use The Elasticsearch Producer
> > To: dev-subscr...@samza.apache.org
> > Cc: Andy Chambers <andy.chamb...@fundingcircle.com>, Charles Reese <
> > charles.re...@fundingcircle.com>
> >
> >
> > Hey guys,
> >
> > I'm trying to use the Elasticsearch Producer, as described in this
> message
> > <https://reviews.apache.org/r/36768/diff/1#3>. But I run into an
> > exception,
> > trying to write out the message.
> >
> > 15:15:59.586 [ThreadJob] INFO o.a.samza.container.SamzaContainer -
> > Entering run loop.
> > 15:16:13.998 [ThreadJob] INFO o.a.samza.container.TaskInstance -
> > SystemStreamPartition [kafka, ledger, 0] is catched up.
> > 15:16:14.006 [ThreadJob] INFO ledger.elastic.job -
> > #object[org.apache.samza.container.TaskInstance$$anon$1 0x20904646
> > org.apache.samza.container.TaskInstance$$anon$1@20904646]
> > 15:16:14.011 [ThreadJob] INFO ledger.elastic.job - {:class
> > org.apache.samza.system.IncomingMessageEnvelope, :key nil, :message
> > :qwerty, :offset 5, :size 0, :system-stream-partition {:class
> > org.apache.samza.system.SystemStreamPartition, :partition {:class
> > org.apache.samza.Partition, :partition-id 0}, :stream ledger, :system
> > kafka, :system-stream {:class org.apache.samza.system.SystemStream,
> :stream
> > ledger, :system kafka}}}
> > 15:16:14.016 [ThreadJob] ERROR o.a.samza.container.SamzaContainer -
> Caught
> > exception in process loop.
> > *org.apache.samza.SamzaException: Attempting to produce to unknown
> system:
> > elasticsearch. Available systems: Set(kafka). Please add the system to
> your
> > configuration, or update outgoing message envelope to send to a defined
> > system.*
> > at
> >
> >
> org.apache.samza.system.SystemProducers$$anonfun$2.apply(SystemProducers.scala:86)
> > at
> >
> >
> org.apache.samza.system.SystemProducers$$anonfun$2.apply(SystemProducers.scala:86)
> > ...
> > at
> org.apache.samza.container.SamzaContainer.run(SamzaContainer.scala:553)
> > at org.apache.samza.job.local.ThreadJob$$anon$1.run(ThreadJob.scala:42)
> >
> >
> > But the source of
> > <
> >
> https://github.com/apache/samza/blob/master/samza-core/src/main/scala/org/apache/samza/system/SystemProducers.scala#L86
> > >
> > the stack trace line, tells me that the code can't find a *producer* in
> the
> > elasticsearch system. The code that I'm using is straight out of the
> > example.
> >
> > collector.send(new OutgoingMessageEnvelope(new
> > SystemStream("elasticsearch", ELASTICSEARCH_INDEX + "/" +
> > ELASTICSEARCH_TYPE), parsedJsonObject));
> >
> >
> > And if we print out the config, I'm indeed seeing *elasticsearch* as a
> > system.
> >
> >
> > 14:09:19.978 [main] INFO ledger.run - ;;
> > ;; ___ __ _ _ __ ___ ______ _
> > ;; / __|/ _` | '_ ` _ \|_ / _` |
> > ;; \__ \ (_| | | | | | |/ / (_| |
> > ;; |___/\__,_|_| |_| |_/___\__,_|
> > ;;
> >
> > 14:09:19.980 [main] INFO ledger.run - ;; >>>>>>>
> ledger-elastic-connector
> > <<<<<<<
> > 14:09:19.980 [main] INFO ledger.run - ;;
> > 14:09:19.981 [main] INFO ledger.run - ;;
> > job.coordinator.replication.factor=1
> > 14:09:19.981 [main] INFO ledger.run - ;; job.coordinator.system=kafka
> > 14:09:19.981 [main] INFO ledger.run - ;;
> > job.factory.class=org.apache.samza.job.local.ThreadJobFactory
> > 14:09:19.981 [main] INFO ledger.run - ;; job.name
> > =ledger-elastic-connector
> > 14:09:19.981 [main] INFO ledger.run - ;;
> > samja.task.factory.function=ledger.elastic.job/elastic-connector
> > 14:09:19.981 [main] INFO ledger.run - ;;
> > serializers.registry.edn.class=samja.core.EDNSerdeFactory
> > 14:09:19.981 [main] INFO ledger.run - ;;
> > systems.elasticsearch.client.elasticsearch.cluster.name=elasticsearch
> > 14:09:19.982 [main] INFO ledger.run - ;;
> >
> >
> systems.elasticsearch.client.factory=org.apache.samza.system.elasticsearch.clientTransport.ClientFactory
> > 14:09:19.982 [main] INFO ledger.run - ;;
> > systems.elasticsearch.client.transport.host=localhost
> > 14:09:19.982 [main] INFO ledger.run - ;;
> > systems.elasticsearch.client.transport.port=9300
> > 14:09:19.982 [main] INFO ledger.run - ;;
> >
> >
> systems.elasticsearch.index.request.factory=org.apache.samza.system.elasticsearch.indexrequest.DefaultIndexRequestFactory
> > 14:09:19.982 [main] INFO ledger.run - ;;
> >
> >
> systems.elasticsearch.samza.factory=org.apache.samza.system.elasticsearch.ElasticsearchSystemFactory
> > 14:09:19.982 [main] INFO ledger.run - ;;
> > systems.kafka.consumer.zookeeper.connect=192.168.99.100:2181
> > 14:09:19.982 [main] INFO ledger.run - ;;
> > systems.kafka.producer.bootstrap.servers=192.168.99.100:9092
> > 14:09:19.982 [main] INFO ledger.run - ;;
> >
> >
> systems.kafka.samza.factory=org.apache.samza.system.kafka.KafkaSystemFactory
> > 14:09:19.982 [main] INFO ledger.run - ;;
> systems.kafka.samza.key.serde=edn
> > 14:09:19.982 [main] INFO ledger.run - ;;
> systems.kafka.samza.msg.serde=edn
> > 14:09:19.982 [main] INFO ledger.run - ;;
> >
> >
> task.checkpoint.factory=org.apache.samza.checkpoint.kafka.KafkaCheckpointManagerFactory
> > 14:09:19.982 [main] INFO ledger.run - ;;
> > task.checkpoint.replication.factor=1
> > 14:09:19.982 [main] INFO ledger.run - ;; task.checkpoint.system=kafka
> > 14:09:19.982 [main] INFO ledger.run - ;;
> task.class=samja.task.SimpleTask
> > 14:09:19.983 [main] INFO ledger.run - ;;
> >
> >
> task.inputs=kafka.ledger,kafka.cash-movement-performed,kafka.ledger-entry-added
> >
> >
> > Is there any other setup, or switches I need to make, to be able to
> > successfully write a message out to elasticsearch?
> >
> >
> > Thanks
> >
> > *Tim Washington* | *Senior Software Engineer*
> >
> > *tim.washing...@fundingcircle.com <renee.nich...@fundingcircle.com>*
> >
> > 747 Front St, 4th Fl | San Francisco, CA 94111
> >
> > *Our Mission: **T**o build a better financial world*
> >
>
