Add ack and compression parameters for Kafka #1359 #1712

chandresh-pancholi · 2019-08-06T09:38:00Z

Which problem is this PR solving?

Resolves Improve configuration parameters for Kafka #1359

Short description of the changes

Add RequiredAck, Compressor, and Compression Level params

chandresh-pancholi · 2019-08-06T10:14:43Z

@yurishkuro, I have made all the required changes along with the test. Kindly review.

yurishkuro · 2019-08-09T03:04:55Z

plugin/storage/kafka/options.go

+	defaultRequiredAcks     = "local"
+	defaultCompression      = "none"
+	defaultCompressionLevel = -1000
+	suffixProtocolVersion   = ".protocol-version"


please keep suffix constants together

yurishkuro · 2019-08-09T03:09:51Z

plugin/storage/kafka/options.go

+func getCompressionLevel(compressionMode string, compressionLevel int) (int, error) {
+	compressionModeData := compressionModes[compressionMode]
+
+	if compressionLevel == defaultCompressionLevel {


this is confusing, I suggest if compressionLevel == 0 {, which makes it obvious that default type's default value is being converted.

This is a snippet from sarama library. this is the reason why I made defaultCompressionLevel = -1000

// CompressionLevelDefault is the constant to use in CompressionLevel // to have the default compression level for any codec. The value is picked // that we don't use any existing compression levels. const CompressionLevelDefault = -1000

that's not the point. The meaning of "default" in sarama is different from what we have here. When dealing with CLI, if you want to use the default value for the respective compressor you simply not specify compression-level flag. Both 0 and -1000 are invalid values for most compressors except for zstd, where BOTH are valid and neither is the default. So I don't see a reason why this bad parameter design from sarama should be carried over to our flags. 0 is the natural null value for int field. It makes no sense to be playing tricks with -1000, which achieve the same results.

I agree with you, i will make the changes

It's been fixed.

i prefer if we compare with 0 explicitly, not with defaultCompressionLevel which happens to be 0, because it makes it easier to understand the logic (you're replacing the zero value).

yurishkuro · 2019-08-09T03:10:39Z

plugin/storage/kafka/options.go

+
+// getCompressionLevel to get compression level from compression type
+func getCompressionLevel(compressionMode string, compressionLevel int) (int, error) {
+	compressionModeData := compressionModes[compressionMode]


in the other functions you use ToLower, should it be consistent?

yurishkuro · 2019-08-09T03:12:04Z

plugin/storage/kafka/options.go

+		return compressionModeData.defaultCompressionLevel, nil
+	}
+
+	if compressionModeData.minCompressionLevel <= compressionLevel && compressionModeData.maxCompressionLevel >= compressionLevel {


I suggest reversing the condition and returning an error "compression level is not within valid range"

yurishkuro · 2019-08-09T03:13:57Z

plugin/storage/kafka/options_test.go

+	assert.Equal(t, val.compressor, sarama.CompressionSnappy)
+}
+
+func TestRequiredAcks(t *testing.T) {


I don't think this test is useful, it just tests that map[] works as expected

fixed how? what is it testing? It's just a map lookup.

My bad. I have removed it now.

yurishkuro · 2019-08-09T03:14:04Z

plugin/storage/kafka/options_test.go

+	require.Error(t, err)
+}
+
+func TestCompressionModes(t *testing.T) {


I don't think this test is useful, it just tests that map[] works as expected

plugin/storage/kafka/options_test.go

yurishkuro · 2019-08-10T21:32:58Z

plugin/storage/kafka/options.go

+	}
+
+	if compressionModeData.minCompressionLevel > compressionLevel || compressionModeData.maxCompressionLevel < compressionLevel {
+		return 0, fmt.Errorf("compression level is not within valid range")


could make the error more descriptive:

fmt.Errorf("compression level %d for '%s' is not within valid range [%d, %d]", level, mode, min, max)

yurishkuro · 2019-08-10T21:35:03Z

plugin/storage/kafka/options.go

+}
+
+//getCompressionModes maps input modes to sarama CompressionCodec
+func getCompressionModes(mode string) (sarama.CompressionCodec, error) {


s/getCompressionModes/getCompressionMode/ (it returns a singular value, not plural)

yurishkuro · 2019-08-10T21:37:16Z

plugin/storage/kafka/options_test.go

+
+func TestCompressionModesFailures(t *testing.T) {
+	_, err := getCompressionModes("test")
+	require.Error(t, err)


nit: this should be assert, not require (you CAN continue the test)

codecov · 2019-08-20T11:12:04Z

Codecov Report

❗ No coverage uploaded for pull request base (master@2f98e82). Click here to learn what that means.
The diff coverage is 87.23%.

@@            Coverage Diff            @@
##             master    #1712   +/-   ##
=========================================
  Coverage          ?   98.17%           
=========================================
  Files             ?      195           
  Lines             ?     9602           
  Branches          ?        0           
=========================================
  Hits              ?     9427           
  Misses            ?      137           
  Partials          ?       38

Impacted Files	Coverage Δ
plugin/storage/kafka/options.go	`92.2% <87.23%> (ø)`

Continue to review full report at Codecov.

Legend - Click here to learn more
Δ = absolute <relative> (impact), ø = not affected, ? = missing data
Powered by Codecov. Last update 2f98e82...e5582d9. Read the comment docs.

chandresh-pancholi · 2019-08-26T11:11:13Z

@yurishkuro Kindly review.

objectiser · 2019-09-05T09:27:17Z

plugin/storage/kafka/options.go

+	suffixEncoding         = ".encoding"
+	suffixRequiredAcks     = ".required-acks"
+	suffixCompression      = ".compression"
+	suffixCompressionLevel = ".compression.level"


Could you change this to compression-level otherwise it will cause problems when specifying the options in yaml due to the previous option.

plugin/storage/kafka/options.go

objectiser · 2019-09-05T10:22:24Z

plugin/storage/kafka/options_test.go

+	_, err := getCompressionMode("test")
+	assert.Error(t, err)
+
+	_, err = getCompressionMode("temp")


Why testing test and temp - isn't this one redundant?

objectiser · 2019-09-05T10:22:59Z

plugin/storage/kafka/options_test.go

+	assert.Error(t, err)
+}
+
+func TestRequiredAcksFailures(t *testing.T) {


What about testing the valid values aswell?

Also same comment as above related to test and temp?

fixed it. Also added test for valid values.

yurishkuro · 2019-09-06T19:48:25Z

NB: the code coverage drop is only from error clauses.

chandresh-pancholi · 2019-09-10T10:19:25Z

@yurishkuro How do you propose to test it? I am not able to figure it out.

objectiser · 2019-09-10T14:24:03Z

plugin/storage/kafka/options_test.go

+		"--kafka.producer.encoding=protobuf",
+		"--kafka.producer.required-acks=local",
+		"--kafka.producer.compression=gzip",
+		"--kafka.producer.compression.level=6"})


This should have been changed to compression-level - so not sure why the unit tests seems to be passing atm?

Such a stupid mistake from my side. :(
fixed and pushed it.

objectiser · 2019-09-11T10:38:06Z

plugin/storage/kafka/options_test.go

+		"--kafka.producer.encoding=protobuf",
+		"--kafka.producer.required-acks=local",
+		"--kafka.producer.compression=gzip",
+		"--kafka.producer.compression-level=6"})


You need to change the value, as 6 is the default for gzip - so regardless of whether this flag is present, the test would currently pass.

Signed-off-by: chandresh-pancholi <[email protected]>

objectiser

@chandresh-pancholi thanks.

For reference in the future, if you could avoid squashing commits during the review cycles, it makes it easier to review changes. The commits will then be squashed when the PR is merged.

objectiser · 2019-09-11T16:29:02Z

@chandresh-pancholi To address the code coverage, you should create some additional tests, variations of TestOptionsWithFlags, where the flags exhibit some error values.

chandresh-pancholi · 2019-09-11T16:29:39Z

@objectiser Thank you so much for the review. I didn't know about the squashing commit. I will keep that in mind in future commits.

chandresh-pancholi · 2019-09-11T16:34:58Z

@objectiser I agree but how to test log.Fatal? It's easy to reach to that line but assertion won't help here. What's your suggestion? I think of below code just to increase the code coverage.

func TestOptionsWithFlagsWithInvalidRequiredAcks(t *testing.T)  {
	opts := &Options{}
	v, command := config.Viperize(opts.AddFlags)
	command.ParseFlags([]string{
		"--kafka.producer.topic=topic1",
		"--kafka.producer.brokers=127.0.0.1:9092, 0.0.0:1234",
		"--kafka.producer.encoding=protobuf",
		"--kafka.producer.required-acks=test",
		"--kafka.producer.compression=gzip",
		"--kafka.producer.compression-level=7"})
	opts.InitFromViper(v)
}

chandresh-pancholi · 2019-09-11T16:38:00Z

is this PR mergable to master?

objectiser · 2019-09-11T18:52:23Z

@chandresh-pancholi Good point - possibly the best way would be to return the error and handle up the stack? From quick look the other storage plugins don't seem to validate the options there.

@yurishkuro Thoughts?

jpkrohling

LGTM. The coverage for the error handling inside InitFromViper could be fixed with some hacking, but not sure it brings much value, as the underlying error conditions are being tested in isolation already.

yurishkuro · 2019-09-12T20:54:45Z

how to test log.Fatal?

Exactly why we, as a rule, don't use log, except in main's.

yurishkuro · 2019-09-06T19:45:56Z

plugin/storage/kafka/options.go

+func getCompressionLevel(compressionMode string, compressionLevel int) (int, error) {
+	compressionModeData := compressionModes[compressionMode]
+
+	if compressionLevel == defaultCompressionLevel {


i prefer if we compare with 0 explicitly, not with defaultCompressionLevel which happens to be 0, because it makes it easier to understand the logic (you're replacing the zero value).

chandresh-pancholi requested review from black-adder, jpkrohling, objectiser, pavolloffay, tiffon, vprithvi and yurishkuro as code owners August 6, 2019 09:38

chandresh-pancholi closed this Aug 8, 2019

chandresh-pancholi force-pushed the jaeger-1359 branch from 48519aa to 464acd0 Compare August 8, 2019 15:53

chandresh-pancholi reopened this Aug 8, 2019

yurishkuro reviewed Aug 9, 2019

View reviewed changes

chandresh-pancholi force-pushed the jaeger-1359 branch 3 times, most recently from 85e1a37 to 6974ec5 Compare August 9, 2019 14:34

chandresh-pancholi closed this Aug 9, 2019

chandresh-pancholi force-pushed the jaeger-1359 branch from 725f179 to 464acd0 Compare August 9, 2019 15:10

chandresh-pancholi reopened this Aug 9, 2019

chandresh-pancholi closed this Aug 9, 2019

chandresh-pancholi force-pushed the jaeger-1359 branch from 56b9891 to 464acd0 Compare August 9, 2019 15:20

chandresh-pancholi reopened this Aug 9, 2019

yurishkuro reviewed Aug 10, 2019

View reviewed changes

chandresh-pancholi force-pushed the jaeger-1359 branch from 7c410c9 to f43a8cd Compare August 20, 2019 11:12

chandresh-pancholi force-pushed the jaeger-1359 branch 2 times, most recently from 505a16c to 2491996 Compare August 20, 2019 11:49

objectiser reviewed Sep 5, 2019

View reviewed changes

chandresh-pancholi closed this Sep 10, 2019

chandresh-pancholi force-pushed the jaeger-1359 branch from 94e6e9b to 464acd0 Compare September 10, 2019 10:57

chandresh-pancholi reopened this Sep 10, 2019

objectiser reviewed Sep 10, 2019

View reviewed changes

chandresh-pancholi force-pushed the jaeger-1359 branch 2 times, most recently from 0c22184 to 8bd740b Compare September 11, 2019 06:41

objectiser reviewed Sep 11, 2019

View reviewed changes

chandresh-pancholi closed this Sep 11, 2019

chandresh-pancholi force-pushed the jaeger-1359 branch from 8bd740b to 464acd0 Compare September 11, 2019 15:07

chandresh-pancholi reopened this Sep 11, 2019

Improve configuration parameters for Kafka jaegertracing#1359

e5582d9

Signed-off-by: chandresh-pancholi <[email protected]>

chandresh-pancholi force-pushed the jaeger-1359 branch from 90e43d2 to e5582d9 Compare September 11, 2019 15:10

objectiser approved these changes Sep 11, 2019

View reviewed changes

jpkrohling approved these changes Sep 12, 2019

View reviewed changes

yurishkuro approved these changes Sep 12, 2019

View reviewed changes

yurishkuro changed the title ~~Improve configuration parameters for Kafka #1359~~ Add ack and compression parameters for Kafka #1359 Sep 12, 2019

yurishkuro merged commit 865f169 into jaegertracing:master Sep 12, 2019

yurishkuro added this to the Release 1.15 milestone Nov 7, 2019

Add ack and compression parameters for Kafka #1359 #1712

Add ack and compression parameters for Kafka #1359 #1712

Conversation

chandresh-pancholi commented Aug 6, 2019 • edited by yurishkuro Loading

Which problem is this PR solving?

Short description of the changes

chandresh-pancholi commented Aug 6, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yurishkuro Aug 10, 2019 • edited Loading

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

codecov bot commented Aug 20, 2019 • edited Loading

Codecov Report

chandresh-pancholi commented Aug 26, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

yurishkuro commented Sep 6, 2019

chandresh-pancholi commented Sep 10, 2019

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

objectiser left a comment

Choose a reason for hiding this comment

objectiser commented Sep 11, 2019

chandresh-pancholi commented Sep 11, 2019

chandresh-pancholi commented Sep 11, 2019

chandresh-pancholi commented Sep 11, 2019

objectiser commented Sep 11, 2019

jpkrohling left a comment

Choose a reason for hiding this comment

yurishkuro commented Sep 12, 2019

Choose a reason for hiding this comment

chandresh-pancholi commented Aug 6, 2019 •

edited by yurishkuro

Loading

yurishkuro Aug 10, 2019 •

edited

Loading

codecov bot commented Aug 20, 2019 •

edited

Loading