print timestamp as millis; fixup from prior commit

- for formatting the record timestamp, we should format from its millis,
because record timestamps are in milliseconds

Author: twmb

commands/consume/transaction_state.go

@@ -35,8 +35,8 @@ func (co *consumeOutput) formatTransactionState(out []byte, r *kgo.Record) []byt
 
 	var keep bool
 	switch v := r.Key[1]; v {
-	case 0, 1:
-		out, keep = co.formatTransactionStateV01(out, r, v)
+	case 0:
+		out, keep = co.formatTransactionStateV0(out, r)
 	default:
 		out = append(out, "(unknown transaction state key format version "...)
 		out = append(out, r.Key[1])
@@ -50,7 +50,7 @@ func (co *consumeOutput) formatTransactionState(out []byte, r *kgo.Record) []byt
 	return out
 }
 
-func (co *consumeOutput) formatTransactionStateV01(dst []byte, r *kgo.Record, version byte) ([]byte, bool) {
+func (co *consumeOutput) formatTransactionStateV0(dst []byte, r *kgo.Record) ([]byte, bool) {
 	{
 		var k kmsg.TxnMetadataKey
 		if err := k.ReadFrom(r.Key); err != nil {

commands/produce/produce.go

@@ -151,7 +151,7 @@ Unfortunately, with exact sizing, the format string is unavoidably noisy.
 				out.Die("cannot produce to a specific topic; the parse format specifies that it parses a topic")
 			}
 
-			var verboseFn func([]byte, *kgo.Record) []byte
+			var verboseFn func([]byte, *kgo.Record, *kgo.FetchPartition) []byte
 			var verboseBuf []byte
 			if verboseFormat != "" {
 				verboseFn, err = format.ParseWriteFormat(verboseFormat, escape)
@@ -196,6 +196,7 @@ Unfortunately, with exact sizing, the format string is unavoidably noisy.
 				cl.AddOpt(kgo.ProduceRetries(retries))
 			}
 
+			p := &kgo.FetchPartition{}
 			for {
 				r, err := reader.Next()
 				if err != nil {
@@ -207,14 +208,13 @@ Unfortunately, with exact sizing, the format string is unavoidably noisy.
 				if !reader.ParsesTopic() {
 					r.Topic = args[0]
 				}
-				err = cl.Client().Produce(context.Background(), r, func(r *kgo.Record, err error) {
+				cl.Client().Produce(context.Background(), r, func(r *kgo.Record, err error) {
 					out.MaybeDie(err, "unable to produce record: %v", err)
 					if verboseFn != nil {
-						verboseBuf = verboseFn(verboseBuf[:0], r)
+						verboseBuf = verboseFn(verboseBuf[:0], r, p)
 						os.Stdout.Write(verboseBuf)
 					}
 				})
-				out.MaybeDie(err, "unable to produce record: %v", err)
 			}
 
 			cl.Client().Flush(context.Background())

commands/transact/transact.go

@@ -199,7 +199,7 @@ func transact(
 	quitCtx context.Context,
 	cl *kgo.Client,
 	txnSess *kgo.GroupTransactSession,
-	w func([]byte, *kgo.Record) []byte,
+	w func([]byte, *kgo.Record, *kgo.FetchPartition) []byte,
 	r *format.Reader,
 	destTopic string,
 	verbose bool,
@@ -228,7 +228,7 @@ func transact(
 				for _, partition := range topic.Partitions {
 					out.MaybeDie(partition.Err, "fetch partition error: %v", partition.Err)
 					for _, record := range partition.Records {
-						buf = w(buf, record)
+						buf = w(buf, record, &partition)
 					}
 				}
 			}

format/parse_out.go

@@ -208,13 +208,14 @@ func ParseWriteFormat(format string, escape rune) (func([]byte, *kgo.Record, *kg
 							return nil, fmt.Errorf("unknown %sd{ time specification", escstr)
 						}
 						format = format[n:]
+
 						argFns = append(argFns, func(out []byte, r *kgo.Record, _ *kgo.FetchPartition) []byte {
-							return numfn(out, int64(r.Timestamp.UnixNano()))
+							return numfn(out, int64(r.Timestamp.UnixNano())/1e6)
 						})
 					}
 				} else {
 					argFns = append(argFns, func(out []byte, r *kgo.Record, _ *kgo.FetchPartition) []byte {
-						return strconv.AppendInt(out, r.Timestamp.UnixNano(), 10)
+						return strconv.AppendInt(out, r.Timestamp.UnixNano()/1e6, 10)
 					})
 				}