KAFKA-5660: Don't throw TopologyBuilderException during runtime #4605

nafshartous · 2018-02-20T22:49:49Z

TopologyBuilderException is a pre-runtime exception that should only be thrown before KafkaStreams#start() is called.

mjsax · 2018-02-20T23:51:28Z

\cc @bbejeck

mjsax

Thanks for the PR. Couple of comments.

@guozhangwang Do we need a KIP for this?

mjsax · 2018-02-21T03:20:27Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamJoinWindow.java

     */
    KStreamJoinWindow(String windowName, long windowSizeMs, long retentionPeriodMs) {
        this.windowName = windowName;

        if (windowSizeMs > retentionPeriodMs)
-            throw new TopologyException("The retention period of the join window "
+            throw new StreamsException("The retention period of the join window "


I think, that a TopologyException is correct here. KStreamJoinWindows is called before KafkaStreams#start(). Or do I miss anything?

mjsax · 2018-02-21T03:24:49Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java

@@ -60,7 +60,7 @@ public RecordCollector recordCollector() {
    @Override
    public StateStore getStateStore(final String name) {
        if (currentNode() == null) {
-            throw new org.apache.kafka.streams.errors.TopologyBuilderException("Accessing from an unknown node");
+            throw new org.apache.kafka.streams.errors.StreamsException("Accessing from an unknown node");


We can import StreamsException an get rid of the package prefix.

(We just did not import TopologyBuilderException because it's a deprecated class and for imports we cannot suppress warnings about usage of deprecated code.

Btw: you can also remove @SuppressWarning from this method.

mjsax · 2018-02-21T03:25:42Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java

@@ -69,7 +69,7 @@ public StateStore getStateStore(final String name) {
        }

        if (!currentNode().stateStores.contains(name)) {
-            throw new org.apache.kafka.streams.errors.TopologyBuilderException("Processor " + currentNode().name() + " has no access to StateStore " + name);
+            throw new org.apache.kafka.streams.errors.StreamsException("Processor " + currentNode().name() + " has no access to StateStore " + name);


mjsax · 2018-02-21T03:25:46Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/StreamPartitionAssignor.java

@@ -644,7 +644,7 @@ private void prepareTopic(final Map<String, InternalTopicMetadata> topicPartitio
                continue;
            }
            if (numPartitions < 0) {
-                throw new org.apache.kafka.streams.errors.TopologyBuilderException(String.format("%sTopic [%s] number of partitions not defined", logPrefix, topic.name()));
+                throw new org.apache.kafka.streams.errors.StreamsException(String.format("%sTopic [%s] number of partitions not defined", logPrefix, topic.name()));


mjsax · 2018-02-21T03:26:55Z

streams/src/test/java/org/apache/kafka/streams/TopologyTest.java

@@ -157,7 +156,7 @@ public void testNamedTopicMatchesAlreadyProvidedPattern() {
    }

    @Test
-    public void shouldNotAllowToAddProcessorWithSameName() {
+    public void shoudNotAllowToAddProcessorWithSameName() {


I think this is a typo that can be reverted.

mjsax · 2018-02-21T03:30:32Z

streams/src/test/java/org/apache/kafka/streams/TopologyTest.java

-                throw new RuntimeException("Did expect different exception. Did catch:", e);
-            }
-        }
+        new ProcessorTopologyTestDriver(streamsConfig, topology.internalTopologyBuilder);


We should keep try-catch and check the error message. Also add a fail if not exception is thrown.

try { new ProcessorTopologyTestDriver(streamsConfig, topology.internalTopologyBuilder); fail("Should have thrown StreamsException"); } catch (final StreamException e) { ...

mjsax · 2018-02-21T03:30:44Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java

@@ -590,8 +590,7 @@ public void shouldAddInternalTopicConfigForRepartitionTopics() {
        assertEquals("appId-foo", topicConfig.name());
    }

-
-    @Test(expected = TopologyBuilderException.class)
+    @Test(expected = StreamsException.class)


mjsax · 2018-02-21T03:30:58Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java

@@ -600,31 +599,24 @@ public void shouldThroughOnUnassignedStateStoreAccess() throws Exception {
        final Properties config = new Properties();
        config.put(StreamsConfig.BOOTSTRAP_SERVERS_CONFIG, "host:1");
        config.put(StreamsConfig.APPLICATION_ID_CONFIG, "appId");
-        config.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory().getAbsolutePath());
+        config.put(StreamsConfig.STATE_DIR_CONFIG, TestUtils.tempDirectory()
+                .getAbsolutePath());


nit: why this?

mjsax · 2018-02-21T03:31:16Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java

+        final TopologyBuilder builder = new TopologyBuilder();
+        builder.addSource(sourceNodeName, "topic")
+                .addProcessor(goodNodeName, new LocalMockProcessorSupplier(),
+                        sourceNodeName)


nit: why this line break?

mjsax · 2018-02-21T03:32:30Z

.../src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java

@@ -560,7 +559,7 @@ public void shouldAddInternalTopicConfigForRepartitionTopics() throws Exception
    }

    @SuppressWarnings("deprecation")
-    @Test
+    @Test(expected = StreamsException.class)


bbejeck · 2018-02-21T20:19:46Z

assuming @mjsax comments addressed, LGTM.

guozhangwang

I don't think we would need a KIP: this is to me more like a bug fix than a public interface change.

guozhangwang · 2018-02-21T21:42:42Z

streams/src/main/java/org/apache/kafka/streams/kstream/internals/KStreamJoinWindow.java

     */
    KStreamJoinWindow(String windowName, long windowSizeMs, long retentionPeriodMs) {
        this.windowName = windowName;

        if (windowSizeMs > retentionPeriodMs)
-            throw new TopologyException("The retention period of the join window "
+            throw new StreamsException("The retention period of the join window "


guozhangwang · 2018-02-21T21:43:38Z

streams/src/test/java/org/apache/kafka/streams/TopologyTest.java

@@ -157,7 +156,7 @@ public void testNamedTopicMatchesAlreadyProvidedPattern() {
    }

    @Test
-    public void shouldNotAllowToAddProcessorWithSameName() {
+    public void shoudNotAllowToAddProcessorWithSameName() {


I think this is a typo that can be reverted.

nafshartous · 2018-02-26T01:35:24Z

Thanks for the comments.

mjsax

Thanks for updating the PR.

Couple more comments.

mjsax · 2018-02-26T19:46:01Z

streams/src/main/java/org/apache/kafka/streams/processor/internals/ProcessorContextImpl.java

@@ -54,13 +55,12 @@ public RecordCollector recordCollector() {
    }

    /**
-     * @throws org.apache.kafka.streams.errors.TopologyBuilderException if an attempt is made to access this state store from an unknown node
+     * @throws org.apache.kafka.streams.errors.StreamsException if an attempt is made to access this state store from an unknown node


Nit: package prefix not needed

mjsax · 2018-02-26T19:48:59Z

streams/src/test/java/org/apache/kafka/streams/TopologyTest.java

-            } else {
-                throw new RuntimeException("Did expect different exception. Did catch:", e);
-            }
+            final String error = e.getCause().toString();


Why do we need to call getCause() ? Does another StreamsException wrap the actual expected StreamsException ? If yes, this should not happen IMHO.

mjsax · 2018-02-26T19:49:13Z

streams/src/test/java/org/apache/kafka/streams/TopologyTest.java

+            final String expectedMessage = "Processor " + badNodeName + " has no access to StateStore " +
+                    LocalMockProcessorSupplier.STORE_NAME;
+
+            assertTrue(error.contains(expectedMessage));


Why not assertEquals ?

Its possible. I'll have to add the fully qualified reference to StreamsException because that's part of the error message.

mjsax · 2018-02-26T19:50:09Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java


+        try {
            final ProcessorTopologyTestDriver driver = new ProcessorTopologyTestDriver(streamsConfig, builder.internalTopologyBuilder);


We should improve this test by moving this line outside/before try

Moving outside the try results in the test failing because this line is throwing the exception

org.apache.kafka.streams.errors.StreamsException: failed to initialize processor badGuy

I see. Than please remove the next line driver.process and simplify to

new ProcessorTopology(...)

(we don't need variable driver for this case)

mjsax · 2018-02-26T19:50:30Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java


+        try {
            final ProcessorTopologyTestDriver driver = new ProcessorTopologyTestDriver(streamsConfig, builder.internalTopologyBuilder);
            driver.process("topic", null, null);
        } catch (final StreamsException e) {


I think we need to insert a fail here to fix this test

mjsax · 2018-02-26T19:50:44Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java

-            } else {
-                throw new RuntimeException("Did expect different exception. Did catch:", e);
-            }
+            final String error = e.getCause().toString();


mjsax · 2018-02-26T19:51:40Z

streams/src/test/java/org/apache/kafka/streams/processor/TopologyBuilderTest.java

+            final String error = e.getCause().toString();
+            final String expectedMessage = "Processor " + badNodeName + " has no access to StateStore ";
+
+            assert error.contains(expectedMessage);


we should not use Java keyword assert but Junit assertXXX methods (assertEquals for this test)

mjsax · 2018-02-26T19:51:55Z

.../src/test/java/org/apache/kafka/streams/processor/internals/InternalTopologyBuilderTest.java

-            }
+            final String expectedMessage = "Processor " + badNodeName + " has no access to StateStore ";
+
+            assertTrue(cause.getMessage().contains(expectedMessage));


nafshartous · 2018-03-01T22:36:03Z

I attempted a merge to my fork to try and resolve the conflict.

git remote add upstream https://github.com/apache/kafka
git fetch upstream
git merge upstream/master

and got

fatal: upstream/master - not something we can merge

mjsax · 2018-03-02T07:29:11Z

Can you try to rebase your PR instead of merging?

nafshartous · 2018-03-02T22:33:21Z

git remote -v
origin	https://github.com/nafshartous/kafka (fetch)
origin	https://github.com/nafshartous/kafka (push)
upstream	git://github.com/apache/kafka.git (fetch)
upstream	git://github.com/apache/kafka.git (push)

git fetch upstream seemed to work. Then I tried git rebase upstream/master and got

fatal: Needed a single revision
invalid upstream upstream/master

mjsax · 2018-03-02T23:16:42Z

Not sure... If you can't clean it up, maybe try to squash all your commits, create a new branch from trunk and cherry pick ?

KAFKA-5660 #comment Don't throw TopologyBuilderException during runtime

3ea52aa

mjsax requested review from dguy, mjsax and guozhangwang February 20, 2018 23:51

mjsax added the streams label Feb 20, 2018

mjsax reviewed Feb 21, 2018

View reviewed changes

guozhangwang reviewed Feb 21, 2018

View reviewed changes

KAFKA-5660 #comment Address PR comments.

8d624df

mjsax changed the title ~~KAFKA-5660 Don't throw TopologyBuilderException during runtime~~ KAFKA-5660: Don't throw TopologyBuilderException during runtime Feb 26, 2018

mjsax reviewed Feb 26, 2018

View reviewed changes

KAFKA-5660 Address comments 2nd pass.

9838d6e

nafshartous closed this Mar 3, 2018

nafshartous mentioned this pull request Mar 4, 2018

KAFKA-5660 Don't throw TopologyBuilderException during runtime #4645

Merged


		try {
		final ProcessorTopologyTestDriver driver = new ProcessorTopologyTestDriver(streamsConfig, builder.internalTopologyBuilder);

KAFKA-5660: Don't throw TopologyBuilderException during runtime #4605

KAFKA-5660: Don't throw TopologyBuilderException during runtime #4605

Conversation

nafshartous commented Feb 20, 2018

mjsax commented Feb 20, 2018

mjsax left a 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

bbejeck commented Feb 21, 2018

guozhangwang left a comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

nafshartous commented Feb 26, 2018

mjsax left a 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

nafshartous commented Mar 1, 2018 • edited Loading

mjsax commented Mar 2, 2018

nafshartous commented Mar 2, 2018

mjsax commented Mar 2, 2018

nafshartous commented Mar 1, 2018 •

edited

Loading