New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Jump to bottom

KSQL-508: add metrics interceptors instead of replacing #585

Merged

rodesai merged 7 commits into master from KSQL-508

Jan 9, 2018

Contributor

rodesai commented Jan 4, 2018

Have the physical plan builder add metrics interceptors instead of
replacing the existing ones. It looks at the current setting, and
builds a list of interceptor classes/names out of it. Finally, the
builder appends the metrics collector interceptor to the list.

To test these changes I added an accessor for the underlying stream
config to QueryMetadata. The tests pull the configured interceptors
from the stream config and verify that they include both the client
and the metrics interceptors.

rodesai added 3 commits

January 3, 2018 17:39


          KSQL-508: add metrics interceptors instead of replacing

4fbe023

Have the physical plan builder add metrics interceptors instead of
replacing the existing ones. It looks at the current setting, and
builds a list of interceptor classes/names out of it. Finally, the
builder appends the metrics collector interceptor to the list.

To test these changes I added an accessor for the underlying stream
config to QueryMetadata. The tests pull the configured interceptors
from the stream config and verify that they include both the client
and the metrics interceptors.


          Fix indentation

7e96cd8


          Unneccessary imports

rodesai requested review from hjafarpour, bluemonk3y and dguy

January 4, 2018 02:50

dguy reviewed

View reviewed changes

ksql-engine/src/main/java/io/confluent/ksql/physical/PhysicalPlanBuilder.java Outdated

+                  List valueList;
+                  if (obj instanceof String) {
+                    String asString = (String)obj;
+                    valueList = new LinkedList<>(Arrays.asList(asString));

Contributor

dguy Jan 4, 2018

why Arrays.asList(...) when you have a string?

Contributor Author

rodesai Jan 4, 2018

The setting accepts a comma-separated list, and there was a .split("\s*,\s*") that got dropped somehow when I was moving some code around. Fixing.

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                  Object val = config.originals().get(
+                      StreamsConfig.consumerPrefix(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);

Contributor

dguy Jan 4, 2018

assertThat(val, instanceOf(List.class))

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                  val = config.originals().get(
+                      StreamsConfig.producerPrefix(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);

Contributor

dguy Jan 4, 2018

as above

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                      StreamsConfig.producerPrefix(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);
+                  List<String> producerInterceptors = (List<String>) val;
+                  Assert.assertEquals(1, producerInterceptors.size());

Contributor

dguy Jan 4, 2018

next 2 lines can be replaced with assertThat(expectedList, equalTo(actualList))

Contributor Author

rodesai Jan 4, 2018

Had something like that originally. Felt like the current check was more robust. We're checking that the name passed to the class loader results in our interceptor class being loaded.

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                  StreamsConfig config = queryMetadata.getStreamsConfig();
+                  Object val = config.originals().get(
+                      StreamsConfig.consumerPrefix(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);

Contributor

dguy Jan 4, 2018

as above

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                      StreamsConfig.consumerPrefix(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);
+                  consumerInterceptors = (List<String>) val;
+                  Assert.assertEquals(2, consumerInterceptors.size());

Contributor

dguy Jan 4, 2018

as above

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                  val = config.originals().get(
+                      StreamsConfig.producerPrefix(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);

Contributor

dguy Jan 4, 2018

as above

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                      StreamsConfig.producerPrefix(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);
+                  producerInterceptors = (List<String>) val;
+                  Assert.assertEquals(2, producerInterceptors.size());

Contributor

dguy Jan 4, 2018

as above

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                      StreamsConfig.consumerPrefix(ConsumerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);
+                  List<String> consumerInterceptors = (List<String>) val;
+                  Assert.assertEquals(2, consumerInterceptors.size());

Contributor

dguy Jan 4, 2018

as above

ksql-engine/src/test/java/io/confluent/ksql/physical/PhysicalPlanBuilderTest.java Outdated

+                      StreamsConfig.producerPrefix(ProducerConfig.INTERCEPTOR_CLASSES_CONFIG));
+                  Assert.assertTrue(val instanceof List);
+                  List<String> producerInterceptors = (List<String>) val;
+                  Assert.assertEquals(2, producerInterceptors.size());

Contributor

dguy Jan 4, 2018

as above

bluemonk3y reviewed

View reviewed changes

bluemonk3y left a comment

only 1 question really as you will see below. The change is trivial however why does it get passed around to the Meta objects? Its its for testing then it should be contained and tested in the owning class and not passed through APIs. This would reduce the amount of code changes required. (Unless I'm missing something!)

ksql-engine/src/main/java/io/confluent/ksql/util/PersistentQueryMetadata.java Outdated

		@@ -34,6 +35,7 @@


		public PersistentQueryMetadata(final String statementString,
		final StreamsConfig streamsConfig,

bluemonk3y Jan 4, 2018

Why does the streamsConfig need to be passed around? Its an internal config state. I can only see it being used in tests - in which case it shouldn't be passed around.

Contributor Author

rodesai Jan 4, 2018

Its just for testing. It was a style choice between making the streams config for a query readable or exposing the function that builds it. Now that I think about it a bit more it would be cleaner to just pull constructing the KafkaStreams out into its own interface/impl that I can mock.

ksql-engine/src/main/java/io/confluent/ksql/util/QueryMetadata.java Outdated

@@ @@ -63,6 +67,10 @@ public String getStatementString() { @@
                   return statementString;
                 }
+                public StreamsConfig getStreamsConfig() {

bluemonk3y Jan 4, 2018

can only see this in the tests... looks like it shouldnt be here...


          Address review comments

99306fa

- Dont pass the stream configs around just for the sake of testing.
  Instead, factor out the last step of KafkaStreams construction into
  a factory interface that we can fake out in the test to intercept
  the args, which includes the streams config
- Split the client interceptor config before turning it into a list

bluemonk3y approved these changes

View reviewed changes

bluemonk3y left a comment

lgtm

ksql-engine/src/main/java/io/confluent/ksql/physical/KafkaStreamsBuilder.java

+              import org.apache.kafka.streams.StreamsBuilder;
+              import org.apache.kafka.streams.StreamsConfig;
+              public interface KafkaStreamsBuilder {

bluemonk3y Jan 5, 2018

This is much better - it would be nice if you could create a fixture or mock that has internal state which is used validate its use; this would be a pita and result in a test thats a bit too functional

Contributor Author

rodesai Jan 5, 2018

I went down that path (using EasyMock) originally but there is so much boilerplate. And its not really adding any more validation than what's being done here. Both approaches validate that KafkaStreamsBuilder is used once and check that the config is consistent with what is expected. I think the only additional thing the mock would do is validate how the mock is used with respect to other mocks, which the test doesn't use anyway.

dguy approved these changes

View reviewed changes

Contributor

dguy left a comment

LGTM, assuming the build passes

bluemonk3y commented Jan 8, 2018

retest this please


          Merge branch 'master' into KSQL-508

f997de1

Contributor Author

rodesai commented Jan 8, 2018

retest this please

satybald reviewed

View reviewed changes

ksql-engine/src/main/java/io/confluent/ksql/physical/PhysicalPlanBuilder.java Outdated

@@ @@ -214,6 +234,23 @@ private String addTimeSuffix(String original) { @@
                   return String.format("%s_%d", original, System.currentTimeMillis());
                 }
+                private void updateListProperty(Map<String, Object> properties, String key, Object value) {
+                  Object obj = properties.getOrDefault(key, new java.util.LinkedList<String>());

Contributor

satybald Jan 8, 2018 •

edited

Loading

I guess there no need to import LL with java.util full path. LinkedList is enough as it already imported.

hjafarpour approved these changes

View reviewed changes

Contributor

hjafarpour left a comment

LGTM.

rodesai added 2 commits

January 9, 2018 13:08


          Merge branch 'master' into KSQL-508

1b42daa


          dont use fully qualified LinkedList name

c3aabf3

rodesai merged commit a9b62ce into master

rodesai deleted the KSQL-508 branch

March 6, 2018 01:03

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Labels

None yet