update transactional produce and consume example
@@ -17,13 +17,17 @@ func startConsuming(ctx context.Context, brokers []string, group, topic string)
return
}
- client.AssignGroup(group, kgo.GroupTopics(topic))
+ client.AssignGroup(group,
+ kgo.GroupTopics(topic),
+
+ // The blocking commit will not run because the only way to
+ // kill this program is to interrupt it, but, usually you will
+ // close the client and wait for it to close before quitting,
+ // and if you want to perform an action on commit errors, you
+ // can use the CommitCallback option.
+ kgo.BlockingCommitOnLeave(),
+ )
defer client.Close()
- // The following commit will not run because the only way to kill this
- // program is to interrupt it, but, usually you should commit the final
- // consumed offsets before quitting, as well as check any commit
- // errors.
- defer func() { client.BlockingCommitOffsets(ctx, client.UncommittedOffsets(), nil) }()
consumerLoop:
for {
