Interesting. Well, i'd say open issue in Sarama's github, probably you'll
find answers/ideas faster.

On Wed, Jul 25, 2018 at 7:51 PM Craig Ching <craigch...@gmail.com> wrote:

> This didn’t fix my problem unfortunately.  Both time stamps are 0.
>
>
> > On Jul 24, 2018, at 15:22, Craig Ching <craigch...@gmail.com> wrote:
> >
> > Hey, thanks for that Dmitriy!  I'll have a look.
> >
> >> On Tue, Jul 24, 2018 at 11:18 AM Dmitriy Vsekhvalnov <
> dvsekhval...@gmail.com> wrote:
> >> Not really associated with Sarama.
> >>
> >> But your issues sounds pretty much same i faced some time ago and fixed,
> >> here it is: https://github.com/Shopify/sarama/issues/885
> >>
> >> Try using msg.BlockTimestamp instead of msg.Timestamp and see if it
> helps.
> >>
> >> On Tue, Jul 24, 2018 at 3:26 AM Craig Ching <craigch...@gmail.com>
> wrote:
> >>
> >> > Hi Dmitry,
> >> >
> >> > Are you associated with the Sarama project?  If so, understand that
> part of
> >> > what I want is to learn about Sarama and the Kafka message format ;)
> >> >
> >> > The problem I'm having is that if I turn on:
> >> >
> >> > log.message.timestamp.type=LogAppendTime
> >> >
> >> > in the broker, then produce on topic1 with console producer, I will
> see
> >> > timestamps in the sarama client.  If I produce on topic2 with telegraf
> >> > (incidentally, I think telegraf is a sarama producer), then I don't
> see
> >> > timestamps in the sarama client.  In both cases, if I consume using
> the
> >> > console consumer (with --property print.timestamp=true) I *do* see
> >> > timestamps.
> >> >
> >> > I'm happy to debug this issue myself and submit a PR to sarama, but I
> am
> >> > missing some fundamentals of how to decode the kafka message format
> and
> >> > would really like some pointers.
> >> >
> >> > Cheers,
> >> > Craig
> >> >
> >> > P.S.  Here is the sarama code I'm using to test:
> >> >
> >> > package main
> >> >
> >> > import (
> >> > "fmt"
> >> > "log"
> >> > "os"
> >> > "os/signal"
> >> > "time"
> >> >
> >> > "github.com/Shopify/sarama"
> >> > )
> >> >
> >> > func main() {
> >> >
> >> > // Initialize Sarama logging
> >> > sarama.Logger = log.New(os.Stdout, "[Sarama] ",
> >> > log.Ldate|log.Lmicroseconds|log.Lshortfile)
> >> >
> >> > signals := make(chan os.Signal, 1)
> >> > signal.Notify(signals, os.Interrupt)
> >> >
> >> > config := sarama.NewConfig()
> >> > config.Consumer.Return.Errors = true
> >> > config.ClientID = "consumer-test"
> >> > config.Metadata.RefreshFrequency = time.Duration(5) * time.Minute
> >> > config.Metadata.Full = true
> >> > // config.Version = sarama.V0_10_0_0
> >> > config.Version = sarama.V1_1_0_0
> >> > // config.Version = sarama.V0_10_2_1
> >> > config.Consumer.Offsets.Initial = sarama.OffsetOldest
> >> >
> >> > brokers := []string{"localhost:9092"}
> >> > // brokers :=
> >> >
> >> >
> []string{"measurement-kafka-broker.service.tgt-pe-prod-ttc.consul.c-prod.ost.cloud.target.internal:9092"}
> >> >
> >> > client, err := sarama.NewConsumer(brokers, config)
> >> > if err != nil {
> >> > panic(err)
> >> > }
> >> >
> >> > // topic := "topic1"
> >> > topic := "topic2"
> >> > // topic := "metric-influx-measurement"
> >> > // How to decide partition, is it fixed value...?
> >> > consumer, err := client.ConsumePartition(topic, 0,
> sarama.OffsetOldest)
> >> > if err != nil {
> >> > panic(err)
> >> > }
> >> >
> >> > defer func() {
> >> > if err := client.Close(); err != nil {
> >> > panic(err)
> >> > }
> >> > }()
> >> >
> >> > // Count how many message processed
> >> > msgCount := 0
> >> >
> >> > go func() {
> >> > for {
> >> > select {
> >> > case err := <-consumer.Errors():
> >> > fmt.Println(err)
> >> > case msg := <-consumer.Messages():
> >> > msgCount++
> >> > fmt.Println(msg.Timestamp)
> >> > fmt.Println("Received messages", string(msg.Key), string(msg.Value))
> >> > case <-signals:
> >> > fmt.Println("Interrupt is detected")
> >> > break
> >> > }
> >> > }
> >> > }()
> >> > <-signals
> >> > }
> >> >
> >> >
> >> > On Mon, Jul 23, 2018 at 10:43 AM Dmitriy Vsekhvalnov <
> >> > dvsekhval...@gmail.com>
> >> > wrote:
> >> >
> >> > > Hey Craig,
> >> > >
> >> > > what exact problem you have with Sarama client?
> >> > >
> >> > > On Mon, Jul 23, 2018 at 5:11 PM Craig Ching <craigch...@gmail.com>
> >> > wrote:
> >> > >
> >> > > > Hi!
> >> > > >
> >> > > > I'm working on debugging a problem with how message timestamps are
> >> > > handled
> >> > > > in the sarama client.  In some cases, the sarama client won't
> >> > associate a
> >> > > > timestamp with a message while the kafka console consumer does.
> I've
> >> > > found
> >> > > > the documentation on the message format here:
> >> > > >
> >> > > > https://kafka.apache.org/documentation/#messageformat
> >> > > >
> >> > > > But the information there is very sparse.  For instance, what are
> >> > > > 'firstTimestamp' and 'maxTimestamp'?  It seems that when I'm
> debugging
> >> > > > sarama, firstTimestamp is set to -1 and maxTimestamp appears to
> be the
> >> > > > timestamp I want.  Is there some state about the message that I
> need to
> >> > > > understand in order to have maxTimestamp be used?  Any further
> >> > > > documentation or guidance on this would be very helpful!
> >> > > >
> >> > > > On another note, I am trying to debug this through the scala/java
> >> > console
> >> > > > consumer, but I'm having a hard time getting IntelliJ setup.  Is
> there
> >> > > > anything special or documentation I need to set this up for
> debugging?
> >> > > >
> >> > >
> >> >
>

Reply via email to