feat(consumer-group): add reset-offset command by rkpattnaik780 · Pull Request #876 · redhat-developer/app-services-cli

rkpattnaik780 · 2021-08-06T11:09:09Z

Implement rhoas kafka consumer-group reset-offset command.

Closes #763

Verification Steps against mock

Start mock server.

make mock-api/start

Select a Kafka instance using

rhoas kafka use --id <id-of-instance>

Run the following command to reset offset for a consumer group:

go run ./cmd/rhoas kafka consumer-group reset-offset --id my-app --offset earliest --topic topic-1

The output should be updated consumers in a tabular form

Verification Steps against stage (WIP)

Create a Kafka instance,a Kafka topic and Kafka consumer group.
After consuming a few messages, disconnect the consumer group.
Run the following command to reset offset for a consumer group:

./rhoas kafka consumer-group reset-offset --id my-group --offset earliest -v

Run the following command to reset offset for specific consumers in a consumer group:

./rhoas kafka consumer-group reset-offset --id my-group --topic rama --offset earliest --partitions "1 2" -v

Type of change

Bug fix (non-breaking change which fixes an issue)
New feature (non-breaking change which adds functionality)
Breaking change (fix or feature that would cause existing functionality to not work as expected)
Documentation change
Other (please specify)

Checklist

Documentation added for the feature
CI and all relevant tests are passing
Code Review completed
Verified independently by reviewer

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

go.mod

bhardesty · 2021-08-09T21:37:14Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Partitions on which reset-offset is to be carried upon (space separated integers)'
+
+[kafka.consumerGroup.resetOffset.flag.output]
+one = 'Output in which to display reset offset result'


Suggested change

one = 'Output in which to display reset offset result'

one = 'Format in which to display reset offset result (choose from: "json", "yml", "yaml")'

bhardesty · 2021-08-09T21:38:47Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Custom offset value (required when offset is absolute or timestamp)'
+
+[kafka.consumerGroup.resetOffset.flag.partitions]
+one = 'Partitions on which reset-offset is to be carried upon (space separated integers)'


Suggested change

one = 'Partitions on which reset-offset is to be carried upon (space separated integers)'

one = 'Partitions on which to reset the consumer group offset (space separated integers)'

bhardesty · 2021-08-09T21:39:28Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Skip confirmation to forcibly reset-offset of a consumer group'
+
+[kafka.consumerGroup.resetOffset.flag.offset]
+one = 'Offset type (earliest, latest, absolute, timestamp)'


Suggested change

one = 'Offset type (earliest, latest, absolute, timestamp)'

one = 'Offset type (choose from: "earliest", "latest", "absolute", "timestamp")'

bhardesty · 2021-08-09T21:41:45Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Select topic for which consumer group offsets are to be reset'
+
+[kafka.consumerGroup.resetOffset.flag.yes]
+one = 'Skip confirmation to forcibly reset-offset of a consumer group'


Suggested change

one = 'Skip confirmation to forcibly reset-offset of a consumer group'

one = 'Skip confirmation to forcibly reset the offset for the consumer group'

bhardesty · 2021-08-09T21:42:00Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

@@ -0,0 +1,43 @@
+[kafka.consumerGroup.resetOffset.cmd.shortDescription]
+one = 'Reset offset of a consumer group'


Suggested change

one = 'Reset offset of a consumer group'

one = 'Reset offset for a consumer group'

bhardesty · 2021-08-09T21:44:23Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+
+[kafka.consumerGroup.resetOffset.cmd.longDescription]
+one = '''
+Reset offset for a consumer group from the Kafka instance.


Suggested change

Reset offset for a consumer group from the Kafka instance.

Reset the offset for a consumer group.

I think the fact that this applies to a Kafka instance should be clear without stating it.

bhardesty · 2021-08-09T21:45:13Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Offset type (earliest, latest, absolute, timestamp)'
+
+[kafka.consumerGroup.resetOffset.flag.value]
+one = 'Custom offset value (required when offset is absolute or timestamp)'


Suggested change

one = 'Custom offset value (required when offset is absolute or timestamp)'

one = 'Custom offset value (required when offset is "absolute" or "timestamp")'

bhardesty · 2021-08-09T21:46:00Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Are you sure you want to reset the offset for consumer group "{{.ID}}"?'
+
+[kafka.consumerGroup.resetOffset.log.debug.cancelledReset]
+one = 'You have chosen to not reset the consumer group offset.'


Suggested change

one = 'You have chosen to not reset the consumer group offset.'

one = 'You have chosen not to reset the consumer group offset.'

bhardesty

Just made a few minor help text suggestions.

craicoverflow · 2021-08-10T10:48:13Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+
+	defer httpRes.Body.Close()
+
+	logger.Info(opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.log.info.successful", localize.NewEntry("ConsumerGroupID", opts.id), localize.NewEntry("InstanceName", kafkaInstance.GetName())))


I wonder should be start breaking long logs onto multiple lines (the code, not the content).

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

craicoverflow · 2021-08-10T10:59:17Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+	localizer  localize.Localizer
+}
+
+type UpdatedConsumerRow struct {


Any reason this was made public?

craicoverflow · 2021-08-10T12:12:44Z

I think it would be better if there was earlier validation of the flag values. For example, if I set offset=absolute but don't pass any value for --value I should be informed of this before a server request is made.

craicoverflow · 2021-08-10T12:13:08Z

Does this require the mocks to test?

rkpattnaik780 · 2021-08-10T12:21:49Z

Does this require the mocks to test?

Yes, I will be updating verification steps.

craicoverflow

I entered a random value for "id" and "topic" but was allowed to go through the steps as if they existed:

❯ ./rhoas kafka consumer-group reset-offset --partitions 1 --partitions 2 --offset latest --id 1 --topic hello
? Are you sure you want to reset the offset for consumer group "1"? Yes
Error: 400 Bad Request. Run the command in verbose mode using the -v flag to see more information

I think we ought to start extracting the error messages from the body:

The error displays "400 Bad Request" but this won't help the user know what to do.

{"code":400,"error_message":"Topic topic1, partition 1 is not valid","class":"IllegalArgumentException"}
Error: 400 Bad Request. Run the command in verbose mode using the -v flag to see more **information**

When I enter an invalid topic, the "Are you sure" dialog gets stuck (hangs). This seems to be an issue on the backend though - the request is sent but it is not responding. I still think we should check if the topic exists on the client-side first.

❯ ./rhoas kafka consumer-group reset-offset --offset earliest --id consumer-group-0 --topic mytopic
? Are you sure you want to reset the offset for consumer group "consumer-group-0"? Yes

Do you think maybe it is a good idea to print the same columns/data that you see in consumer-group describe? There are columns like "log end offset", "current offset" which users may expect to see. (Up for discussion).
Passing an inval;id value for --value is not handled by the API. We might need to log some bugs/work with the team to get this resolved.

./rhoas kafka consumer-group reset-offset \                                  
--id consumer-group-0 \
--topic topic1 -y \
--offset absolute --value "hello" \
-yv

It would be cool to validate the "value" value when the "offset" is "timestamp". The format should be "Timestamp must be in format 'yyyy-MM-dd'T'HH:mm:ssz" but the user cannot tell this from anywhere in the CLI so it needs to be documented.

Overall, good PR but this needs to be tested more extensively as it is an important command with many option combinations possible.

craicoverflow · 2021-08-31T09:03:29Z

pkg/cmd/flag/validation.go

+func ValidateOffset(v string) error {
+	isValid := flagutil.IsValidInput(v, flagutil.ValidOffsets...)
+
+	if isValid {
+		return nil
+	}
+
+	return InvalidValueError("output", v, flagutil.ValidOffsets...)
+}


I see pkg/cmd/flag/validation.go as a place for generic, reusable flag helpers. Offset is specific to consumer groups, and so it should not be here.

Can we have another method with signature ValidateInput(v string, validValues string[], flagName string) ?

Yeah, if you think that would be an in improvement just go ahead and implement it 😄

craicoverflow · 2021-08-31T09:04:43Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+				}
+			}
+
+			if opts.value == "" && (opts.offset == "absolute" || opts.offset == "timestamp") {


Let's create some constants for the different possible values for "offset".

craicoverflow · 2021-08-31T09:14:22Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+	cmd.Flags().BoolVarP(&opts.skipConfirm, "yes", "y", false, opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.yes"))
+	cmd.Flags().StringVar(&opts.id, "id", "", opts.localizer.MustLocalize("kafka.consumerGroup.common.flag.id.description", localize.NewEntry("Action", "reset-offset")))
+	cmd.Flags().StringVar(&opts.value, "value", "", opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.value"))
+	cmd.Flags().StringVar(&opts.offset, "offset", "", opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.offset"))


I wonder if we should be using a default? Does the UI default to anything? From some research I see that it can be configured from Kafka itself, and the default value is "latest". I am not sure whether the backend produces some default.

UI doesn't have a default, renders "select" in the beginning.

craicoverflow · 2021-08-31T09:17:56Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+	switch opts.output {
+	case "json":
+		data, _ := json.Marshal(updatedConsumers)
+		_ = dump.JSON(opts.IO.Out, data)


Use the PrintDataInFormat helper defined in the dump package.

craicoverflow · 2021-08-31T09:18:12Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+	)
+
+	switch opts.output {
+	case "json":


Use the constants defined in dump.

craicoverflow · 2021-08-31T09:18:39Z

pkg/cmdutil/flags/flags.go

 var (
 	ValidOutputFormats       = []string{dump.JSONFormat, dump.YAMLFormat, dump.YMLFormat}
 	CredentialsOutputFormats = []string{"env", "json", "properties"}
+	ValidOffsets             = []string{"timestamp", "absolute", "latest", "earliest"}


Use the constants which will be created :)

craicoverflow · 2021-08-31T09:19:05Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+
+[kafka.consumerGroup.resetOffset.cmd.longDescription]
+one = '''
+Reset the offset for a consumer group.


This should probably have some more detail?

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

craicoverflow · 2021-08-31T09:20:13Z

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go

+	cmd.Flags().StringVar(&opts.value, "value", "", opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.value"))
+	cmd.Flags().StringVar(&opts.offset, "offset", "", opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.offset"))
+	cmd.Flags().StringVar(&opts.topic, "topic", "", opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.topic"))
+	cmd.Flags().Int32SliceVar(&opts.partitions, "partitions", []int32{}, opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.partitions"))


Suggested change

cmd.Flags().Int32SliceVar(&opts.partitions, "partitions", []int32{}, opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.partitions"))

cmd.Flags().Int32SliceVar(&opts.partitions, "partition", []int32{}, opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.partitions"))

I think partitions 1,2,3 will look better than --partition 1 --partition 2 --partition 3. Will become a drag as the number of partitions to pass will increase.

Ah neat, I did not know that Int32SliceVar allowed for both input types to be valid. That is fine then.

craicoverflow · 2021-08-31T09:20:55Z

pkg/localize/locales/en/cmd/kafka_consumergroup_reset_offset.toml

+one = 'Custom offset value (required when offset is "absolute" or "timestamp")'
+
+[kafka.consumerGroup.resetOffset.flag.partitions]
+one = 'Partitions on which to reset the consumer group offset (space separated integers)'


"(space separated integers)" is that right?

If it is right, it should be "space-separated" :-)

It should be comma separated as of now, used to be space separated before.

craicoverflow

A couple of nits, but happy for those to be addressed in a follow up.

craicoverflow · 2021-09-02T12:28:28Z

pkg/kafka/consumergroup/util.go

-	EarliestOffset  = "earliest"
-	TimestampOffset = "timestamp"
-	LatestOffset    = "latest"
+	OffsetAbssolute = "absolute"


Suggested change

OffsetAbssolute = "absolute"

OffsetAbsolute

typo.

craicoverflow · 2021-09-02T12:33:08Z

pkg/kafka/consumergroup/util.go


+var ValidOffsets = []string{OffsetAbssolute, OffsetEarliest, OffsetTimestamp, OffsetLatest}
+
 var timestampOffsetRegExp = regexp.MustCompile(`^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}-\d{2}:\d{2})$`)


It could be better to use an iso8601 validator lib for this:

https://github.com/relvacode/iso8601

Move to validators package.

craicoverflow · 2021-09-02T12:37:33Z

pkg/kafka/consumergroup/util.go

 }
+
+// ValidateOffset checks if value v is a valid value for --offset
+func ValidateOffset(v string) error {


nit: Ideally this should go into the validators package

craicoverflow · 2021-09-02T12:41:04Z

Next steps once this is merged @rkpattnaik780 - add an example to the guides repo.

rkpattnaik780 · 2021-09-02T12:53:28Z

Next steps once this is merged @rkpattnaik780 - add an example to the guides repo.

Will be doing these in a follow up along with tests for consumer group validator.

wtrocki · 2021-09-03T07:06:24Z

Is this feature deployed to prod?

rkpattnaik780 · 2021-09-03T07:13:57Z

Is this feature deployed to prod?

Will check in prod and get back to you, it is there in stage though.

wtrocki · 2021-09-03T07:31:37Z

Please do not bother. I would try to play with this on stage. Going to play with this for a while

craicoverflow · 2021-09-03T08:37:30Z

Is this feature deployed to prod?

I think that it is yes.

wtrocki · 2021-09-03T08:44:47Z

docs/commands/rhoas_kafka_consumer-group_reset-offset.adoc

+[discrete]
+== Synopsis
+
+Reset the offset for consumers in a consumer group reading from a given topic.


We need more information.

List offset types.
We need to list 2 modes (partition mode is only for advanced use cases)

wtrocki · 2021-09-03T08:45:07Z

docs/commands/rhoas_kafka_consumer-group_reset-offset.adoc

+....
+
+[discrete]
+== Examples


Examples should reflect all the modes and types.

rkpattnaik780 · 2021-09-06T17:55:54Z

Next steps once this is merged @rkpattnaik780 - add an example to the guides repo.

Should we create a new release (0.29.0) before updating the guides?

craicoverflow · 2021-09-07T05:58:54Z

Next steps once this is merged @rkpattnaik780 - add an example to the guides repo.

Should we create a new release (0.29.0) before updating the guides?

You should have the pull request ready to go as a draft, and merge it once 0.29.0 is released.

openshift-ci bot added the do-not-merge/work-in-progress label Aug 6, 2021

rkpattnaik780 requested a review from craicoverflow August 6, 2021 11:09

craicoverflow reviewed Aug 6, 2021

View reviewed changes

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go Outdated Show resolved Hide resolved

craicoverflow reviewed Aug 6, 2021

View reviewed changes

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go Outdated Show resolved Hide resolved

rkpattnaik780 force-pushed the reset_offset branch from c0cd238 to 9f22784 Compare August 9, 2021 10:46

rkpattnaik780 requested a review from bhardesty August 9, 2021 14:37

craicoverflow reviewed Aug 9, 2021

View reviewed changes

go.mod Outdated Show resolved Hide resolved

bhardesty reviewed Aug 9, 2021

View reviewed changes

rkpattnaik780 requested review from bhardesty and craicoverflow August 10, 2021 08:26

craicoverflow reviewed Aug 10, 2021

View reviewed changes

pkg/cmd/kafka/consumergroup/resetoffset/reset_offset.go Outdated Show resolved Hide resolved

craicoverflow reviewed Aug 10, 2021

View reviewed changes

rkpattnaik780 requested a review from craicoverflow August 10, 2021 14:21

rkpattnaik780 mentioned this pull request Aug 12, 2021

Replace hardcoded status codes with http constants #896

Closed

rkpattnaik780 marked this pull request as ready for review August 30, 2021 14:12

openshift-ci bot removed the do-not-merge/work-in-progress label Aug 30, 2021

rkpattnaik780 force-pushed the reset_offset branch from 03b7219 to b2c6300 Compare August 30, 2021 14:19

craicoverflow suggested changes Aug 31, 2021

View reviewed changes

rkpattnaik780 added 7 commits September 2, 2021 18:03

fix: address review comments

87761c8

fix: address review comments

49552b8

fix: typo in error message

6c34335

fix: use make for updated consumer row

99568ec

fix: resolve merge conflicts and update logger

5fd0bcf

fix: address review comments

4ccd29c

fix: validator interface for consumer group

9051583

rkpattnaik780 force-pushed the reset_offset branch from 865975a to 9051583 Compare September 2, 2021 12:33

docs: regenrate docs

671e8d8

craicoverflow approved these changes Sep 2, 2021

View reviewed changes

rkpattnaik780 added 2 commits September 2, 2021 18:17

fix: typo AbsoluteOffset

239adba

fix: erroneous switch

a5961c3

rkpattnaik780 requested a review from wtrocki September 2, 2021 12:54

test: add unit tests for offset validations

066519b

wtrocki reviewed Sep 3, 2021

View reviewed changes

wtrocki approved these changes Sep 3, 2021

View reviewed changes

docs: update reset-offset examples

82743fd

rkpattnaik780 merged commit 1b0876f into main Sep 3, 2021

rkpattnaik780 deleted the reset_offset branch September 3, 2021 11:58

rkpattnaik780 added a commit that referenced this pull request Sep 20, 2021

feat(consumer-group): add reset-offset command (#876)

e109dc9

rkpattnaik780 mentioned this pull request Sep 27, 2021

Use validator library to validate timestamp offset value #1147

Closed

	one = 'Output in which to display reset offset result'
	one = 'Format in which to display reset offset result (choose from: "json", "yml", "yaml")'

	one = 'Partitions on which reset-offset is to be carried upon (space separated integers)'
	one = 'Partitions on which to reset the consumer group offset (space separated integers)'

	one = 'Offset type (earliest, latest, absolute, timestamp)'
	one = 'Offset type (choose from: "earliest", "latest", "absolute", "timestamp")'

	one = 'Skip confirmation to forcibly reset-offset of a consumer group'
	one = 'Skip confirmation to forcibly reset the offset for the consumer group'

		@@ -0,0 +1,43 @@
		[kafka.consumerGroup.resetOffset.cmd.shortDescription]
		one = 'Reset offset of a consumer group'

	one = 'Reset offset of a consumer group'
	one = 'Reset offset for a consumer group'

	Reset offset for a consumer group from the Kafka instance.
	Reset the offset for a consumer group.

	one = 'Custom offset value (required when offset is absolute or timestamp)'
	one = 'Custom offset value (required when offset is "absolute" or "timestamp")'

	one = 'You have chosen to not reset the consumer group offset.'
	one = 'You have chosen not to reset the consumer group offset.'


		defer httpRes.Body.Close()

		logger.Info(opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.log.info.successful", localize.NewEntry("ConsumerGroupID", opts.id), localize.NewEntry("InstanceName", kafkaInstance.GetName())))

	cmd.Flags().Int32SliceVar(&opts.partitions, "partitions", []int32{}, opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.partitions"))
	cmd.Flags().Int32SliceVar(&opts.partitions, "partition", []int32{}, opts.localizer.MustLocalize("kafka.consumerGroup.resetOffset.flag.partitions"))


		var ValidOffsets = []string{OffsetAbssolute, OffsetEarliest, OffsetTimestamp, OffsetLatest}

		var timestampOffsetRegExp = regexp.MustCompile(`^(\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}-\d{2}:\d{2})$`)

Conversation

rkpattnaik780 commented Aug 6, 2021 • edited Loading Uh oh! There was an error while loading. Please reload this page.

Uh oh!

Verification Steps against mock

Verification Steps against stage (WIP)

Type of change

Checklist

Uh oh!

Uh oh!

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

bhardesty left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

craicoverflow commented Aug 10, 2021

Uh oh!

craicoverflow commented Aug 10, 2021

Uh oh!

rkpattnaik780 commented Aug 10, 2021

Uh oh!

craicoverflow left a comment

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

Choose a reason for hiding this comment

Uh oh!

craicoverflow left a comment

Choose a reason for hiding this comment

Uh oh!

rkpattnaik780 commented Aug 6, 2021 •

edited

Loading

wtrocki commented Sep 3, 2021 •

edited

Loading