KAFKA-4591: Create Topic Policy (KIP-108) #2361

ijuma · 2017-01-12T21:15:34Z

No description provided.

ijuma · 2017-01-12T21:15:44Z

Will post additional tests soon.

asfbot · 2017-01-12T21:21:22Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/811/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-12T21:58:55Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/809/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-12T22:46:58Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/810/
Test FAILed (JDK 7 and Scala 2.10).

gwenshap

LGTM. Just few nits.

gwenshap · 2017-01-12T22:55:37Z

clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java

-            topicErrors.put(topic, Errors.forException(e));
+            Errors error = Errors.forException(e);
+            // Avoid populating the error message if it's a generic one
+            String message = error.message().equals(e.getMessage()) ? null : e.getMessage();


Hmm... This seems kind of annoying for non-java clients. Is it really worth the savings?

Is it though? Don't they have to map error codes for all requests anyway? The idea was to only include an error message when it provided information over what the error code provides. Do you think sending it back always would be better?

I guess that's fair.

gwenshap · 2017-01-12T23:10:19Z

clients/src/test/java/org/apache/kafka/common/requests/RequestResponseTest.java

-        Map<String, Errors> errors = new HashMap<>();
-        errors.put("t1", Errors.INVALID_TOPIC_EXCEPTION);
-        errors.put("t2", Errors.LEADER_NOT_AVAILABLE);
+        Map<String, CreateTopicsResponse.Error> errors = new HashMap<>();


We don't want to have tests for the two versions?

Yes, working on that and more tests.

gwenshap · 2017-01-12T23:13:13Z

core/src/main/scala/kafka/server/AdminManager.scala

      } catch {
        case e: Throwable =>
-          warn(s"Error processing create topic request for topic $topic with arguments $arguments", e)
-          CreateTopicMetadata(topic, Map(), Errors.forException(e))
+          info(s"Error processing create topic request for topic $topic with arguments $arguments", e)


Why downgrade to info?

This is related to the other comment of yours with regards to logging. To me, warn implies that there's something that needs attention. Invalid topic creation requests are normal in self-serve environments. Does this not make sense?

I guess its a matter of taste :) IMO, it requires attention since it can indicate an unhappy user - maybe reaching out to the user to discuss why he's trying to do what he does? or to explain why he can't? Maybe not all admins are as caring :)

If you are not convinced, leave it as is. We can figure out if admins complain later :)

gwenshap · 2017-01-12T23:18:21Z

core/src/main/scala/kafka/server/KafkaApis.scala

+              s"for the following topics: ${duplicateTopics.keySet.mkString(",")}"
+            // We can send the error message in the response for version 1, so we don't have to log it any more
+            if (request.header.apiVersion == 0)
+              warn(errorMessage)


I think we still need to log. There's the client log and the broker log - IMO, create topic is rare enough and partially an SRE responsibility so the admins will want to monitor for failed create-topics too. Lets give them the option.

Create topics can be somewhat common in self-serve environments. This error only happens if someone sends duplicate topics in their requests. It's not clear to me why a SRE would care about that? It's clearly user error and whoever is creating a topic would just fix it and move on. It would be good to understand from @granthenke what was his reasoning for this warning.

I think I'm with Ismael. Since we return the error message to the user in the new version, logging it on the broker seems redundant.

As a Kafka admin, I'd want it highlighted to my attention. But maybe I'm unique - we can leave it that was and see if anyone else cares :)

gwenshap · 2017-01-12T23:24:23Z

core/src/main/scala/kafka/server/KafkaApis.scala

+            duplicateTopics.keySet.map((_, new CreateTopicsResponse.Error(Errors.INVALID_REQUEST, errorMessage))).toMap
+          } else Map.empty
+
+        val completeResults = results ++ duplicatedTopicsResults
        sendResponseCallback(completeResults)
      }

      adminManager.createTopics(


Not 100% related - but there is a private createTopic method in this file (KafkaApis), and I got a bit confused on why it isn't used. It think it is only for internal topics, but maybe add a comment explaining?

gwenshap · 2017-01-12T23:43:03Z

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

                    continue;
                } else {
-                    throw new StreamsException("Could not create topic: " + internalTopicConfig.name() + ". " + createTopicsResponse.errors().get(internalTopicConfig.name()).name());
+                    throw new StreamsException("Could not create topic: " + internalTopicConfig.name() + " due to " + error.message());


ah, much better!

asfbot · 2017-01-12T23:48:42Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/810/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-12T23:49:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/812/
Test FAILed (JDK 8 and Scala 2.11).

hachikuji · 2017-01-13T19:49:51Z

Mentioned offline, but I thought I would raise it here. I'm not sure InvalidRequestException is ideal for policy validation errors. Its current intended usage is for protocol errors. Maybe we could create a new exception? For example, CreateTopicValidationException?

…pic-policy * apache/trunk: MINOR: Remove unneeded client used API lists MINOR: Cluster file generator should produce valid json KAFKA-4565; Separation of Internal and External traffic (KIP-103) KAFKA-4626; Add KafkaConsumer#close change to upgrade notes KAFKA-4581; Fail early if multiple client login modules in sasl.jaas.config MINOR: Minor improvements in consumer close timeout handling MINOR: Remove unnecessary options in SCRAM test jaas config MINOR: avoid closing over both pre & post-transform record in WorkerSourceTask MINOR: make methods introduced in KAFKA-4490 consistent with KIP-100 KAFKA-3209: KIP-66: single message transforms MINOR: fix JavaDoc in SegmentIterator

…pic-policy * apache/trunk: KAFKA-4381: Add per partition lag metrics to the consumer MINOR: Remove unnecessary store info from TopologyBuilder KAFKA-3739: Add no-arg constructor for WindowedSerdes in Streams HOTFIX: Added another broker to smoke test KAFKA-4627; Fix timing issue in consumer close tests

asfbot · 2017-01-13T20:31:27Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/855/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-13T20:31:31Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/857/
Test FAILed (JDK 8 and Scala 2.11).

hachikuji · 2017-01-13T20:50:49Z

clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java

@@ -167,9 +176,12 @@ private CreateTopicsRequest(Map<String, TopicDetails> topics, Integer timeout, s
        }
        struct.set(REQUESTS_KEY_NAME, createTopicRequestStructs.toArray());
        struct.set(TIMEOUT_KEY_NAME, timeout);
+        if (version >= 1)


If validateOnly is true, I wonder if we should raise an exception if the version is 0. It seems a little dangerous otherwise.

Maybe we can do that in the builder since the constructor is private?

asfbot · 2017-01-13T21:09:17Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/855/
Test FAILed (JDK 7 and Scala 2.10).

apurvam · 2017-01-13T21:11:35Z

Overall looks good to me. I think Gwen and Jason brought up some good minor points, to which I have nothing to add.

ijuma · 2017-01-13T21:27:06Z

About the exception, I'll create a custom exception as @hachikuji suggested. I'll also rename TopicDetails to RequestMetadata to make it a bit more generic for future evolution.

hachikuji · 2017-01-13T21:28:33Z

core/src/main/scala/kafka/server/AdminManager.scala

@@ -98,7 +111,7 @@ class AdminManager(val config: KafkaConfig,
      val results = metadata.map { createTopicMetadata =>
        // ignore topics that already have errors
        if (createTopicMetadata.error == Errors.NONE) {


This check needs to be updated.

Haha, I was trying to diagnose some test failures and reached the exact same conclusion at about the same time. There are quite a few places with the same issue. Will push an update ASAP.

…adata

hachikuji · 2017-01-13T22:05:09Z

core/src/main/scala/kafka/server/KafkaConfig.scala

@@ -488,6 +490,8 @@ object KafkaConfig {
    "create a topic with a replication factor of 3, set min.insync.replicas to 2, and " +
    "produce with acks of \"all\". This will ensure that the producer raises an exception " +
    "if a majority of replicas do not receive a write."
+
+  val CreateTopicsPolicyClassNameDoc = "The create topics policy class that should be used for validation."


Worth mentioning the interface name that users should extend here?

Quite dangerous otherwise.

asfbot · 2017-01-13T22:08:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/862/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-13T22:09:01Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/862/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-13T22:09:38Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/864/
Test FAILed (JDK 8 and Scala 2.11).

hachikuji · 2017-01-13T22:13:09Z

clients/src/main/java/org/apache/kafka/common/requests/CreateTopicsRequest.java

@@ -113,6 +114,8 @@ public Builder(Map<String, TopicDetails> topics, int timeout, boolean validateOn

        @Override
        public CreateTopicsRequest build() {
+            if (validateOnly && version() == 0)


Probably worth a test case for this.

hachikuji · 2017-01-13T22:21:20Z

@ijuma LGTM. I'm ready to merge when you add the additional tests.

asfbot · 2017-01-13T22:34:02Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/867/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-13T22:34:33Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/869/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-13T22:37:14Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/870/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-13T22:37:25Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/868/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-13T22:38:16Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/868/
Test FAILed (JDK 7 and Scala 2.10).

hachikuji · 2017-01-13T22:49:23Z

clients/src/main/java/org/apache/kafka/server/policy/CreateTopicPolicy.java

+
+public interface CreateTopicPolicy {
+
+    class RequestMetadata {


nit: might be nice to have a toString for this.

asfbot · 2017-01-13T23:32:09Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/871/
Test PASSed (JDK 8 and Scala 2.11).

asfbot · 2017-01-13T23:32:49Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/869/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-13T23:32:50Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/869/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-14T01:11:46Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/876/
Test FAILed (JDK 8 and Scala 2.11).

asfbot · 2017-01-14T01:13:32Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/874/
Test FAILed (JDK 7 and Scala 2.10).

asfbot · 2017-01-14T01:14:15Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/874/
Test FAILed (JDK 8 and Scala 2.12).

…pic-policy * apache/trunk: KAFKA-3853; Extend OffsetFetch API to allow fetching all offsets for a consumer group (KIP-88)

asfbot · 2017-01-14T02:55:48Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.11/883/
Test PASSed (JDK 8 and Scala 2.11).

hachikuji

LGTM

asfbot · 2017-01-14T05:09:04Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk8-scala2.12/881/
Test FAILed (JDK 8 and Scala 2.12).

asfbot · 2017-01-14T05:09:31Z

Refer to this link for build results (access rights to CI server needed):
https://builds.apache.org/job/kafka-pr-jdk7-scala2.10/881/
Test FAILed (JDK 7 and Scala 2.10).

Author: Ismael Juma <ismael@juma.me.uk> Reviewers: Apurva Mehta <apurva.1618@gmail.com>, Gwen Shapira <cshapi@gmail.com>, Jason Gustafson <jason@confluent.io> Closes apache#2361 from ijuma/kafka-4591-create-topic-policy

KAFKA-4591: Create Topic Policy (KIP-108)

89736a7

A few fixes

d675e76

gwenshap approved these changes Jan 12, 2017

View reviewed changes

ijuma added 2 commits January 13, 2017 20:19

hachikuji reviewed Jan 13, 2017

View reviewed changes

ijuma added 4 commits January 13, 2017 21:33

Fix types being compared to fix test failures

783e3d1

Rename CreateTopicPolicy.TopicDetails to CreateTopicPolicy.RequestMet…

4f115d0

…adata

Introduce Error.is to make code read better

ee00eab

Introduce PolicyViolationException

db0943b

hachikuji reviewed Jan 13, 2017

View reviewed changes

Throw exception if validateOnly is used in v0 of CreateTopicsRequest

0c17ebf

Quite dangerous otherwise.

hachikuji reviewed Jan 13, 2017

View reviewed changes

Remove unused import

89213b1

ijuma added 2 commits January 13, 2017 22:34

Mention interface name in create.topics.policy.class.name documentation

eb9c7da

Error.toString

eea3762

Fix checkstyle

931cf10

hachikuji reviewed Jan 13, 2017

View reviewed changes

ijuma added 3 commits January 14, 2017 00:06

Fix bug in validateOnly and add integration test

822064c

Add toString to CreateTopicPolicy

48d41d8

Fix policy creation and configuration

90ff284

ijuma added 3 commits January 14, 2017 01:34

Fix serialisation of message and add tests for custom validator

60f4735

Merge remote-tracking branch 'apache/trunk' into kafka-4591-create-to…

98d7e6b

…pic-policy * apache/trunk: KAFKA-3853; Extend OffsetFetch API to allow fetching all offsets for a consumer group (KIP-88)

Missing test changes

e371971

ijuma force-pushed the kafka-4591-create-topic-policy branch from c1a2144 to e371971 Compare January 14, 2017 02:08

hachikuji approved these changes Jan 14, 2017

View reviewed changes

asfgit closed this in da57bc2 Jan 14, 2017

ijuma deleted the kafka-4591-create-topic-policy branch September 5, 2017 09:42

KAFKA-4591: Create Topic Policy (KIP-108) #2361

KAFKA-4591: Create Topic Policy (KIP-108) #2361

Conversation

ijuma commented Jan 12, 2017

ijuma commented Jan 12, 2017

asfbot commented Jan 12, 2017

asfbot commented Jan 12, 2017

asfbot commented Jan 12, 2017

gwenshap 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

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 12, 2017

asfbot commented Jan 12, 2017

hachikuji commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 13, 2017

apurvam commented Jan 13, 2017

ijuma commented Jan 13, 2017

Choose a reason for hiding this comment

Choose a reason for hiding this comment

Choose a reason for hiding this comment

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

Choose a reason for hiding this comment

hachikuji commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

Choose a reason for hiding this comment

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 13, 2017

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017

hachikuji left a comment

Choose a reason for hiding this comment

asfbot commented Jan 14, 2017

asfbot commented Jan 14, 2017