1+ /**
2+ * Licensed to the Apache Software Foundation (ASF) under one or more
3+ * contributor license agreements. See the NOTICE file distributed with
4+ * this work for additional information regarding copyright ownership.
5+ * The ASF licenses this file to You under the Apache License, Version 2.0
6+ * (the "License"); you may not use this file except in compliance with
7+ * the License. You may obtain a copy of the License at
8+ *
9+ * http://www.apache.org/licenses/LICENSE-2.0
10+ *
11+ * Unless required by applicable law or agreed to in writing, software
12+ * distributed under the License is distributed on an "AS IS" BASIS,
13+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
14+ * See the License for the specific language governing permissions and
15+ * limitations under the License.
16+ */
17+
18+ package kafka .server
19+
20+ import java .util .concurrent .TimeUnit
21+ import kafka .network .RequestChannel
22+ import org .apache .kafka .common .requests .{AbstractRequest , AbstractResponse , ProduceRequest , RequestContext }
23+ import org .apache .kafka .common .Configurable
24+
25+ /**
26+ * Top level interface that all pluggable observer must implement. Kafka will read the 'observer.class.name' config
27+ * value at startup time, create an instance of the specificed class using the default constructor, and call its
28+ * 'configure' method.
29+ *
30+ * From that point onwards, every pair of request and response will be routed to the 'record' method.
31+ *
32+ * If 'observer.class.name' has no value specified or the specified class does not exist, the <code>NoOpObserver</code>
33+ * will be used as a place holder.
34+ */
35+ trait Observer extends Configurable {
36+
37+ /**
38+ * Observe a request and its corresponding response
39+ *
40+ * @param requestContext the context information about the request
41+ * @param request the request being observed for a various purpose(s)
42+ * @param response the response to the request
43+ */
44+ def observe (requestContext : RequestContext , request : AbstractRequest , response : AbstractResponse ): Unit
45+
46+ /**
47+ * Observe a produce request. This method handles only the produce request since produce request is special in
48+ * two ways. Firstly, if ACK is set to be 0, there is no produce response associated with the produce request.
49+ * Secondly, the lifecycle of some inner fields in a ProduceRequest is shorter than the lifecycle of the produce
50+ * request itself. That means in some situations, when <code>observe</code> is called on a produce request and
51+ * response pair, some fields in the produce request has been null-ed already so that the produce request and
52+ * response is not observable (or no useful information). Therefore this method exists for the purpose of allowing
53+ * users to observe on the produce request before its corresponding response is created.
54+ *
55+ * @param requestContext the context information about the request
56+ * @param produceRequest the produce request being observed for a various purpose(s)
57+ */
58+ def observeProduceRequest (requestContext : RequestContext , produceRequest : ProduceRequest ): Unit
59+
60+ /**
61+ * Close the observer with timeout.
62+ *
63+ * @param timeout the maximum time to wait to close the observer.
64+ * @param unit the time unit.
65+ */
66+ def close (timeout : Long , unit : TimeUnit ): Unit
67+ }
68+
69+ object Observer {
70+
71+ /**
72+ * Generates a description of the given request and response. It could be used mostly for debugging purpose.
73+ *
74+ * @param request the request being described
75+ * @param response the response to the request
76+ */
77+ def describeRequestAndResponse (request : RequestChannel .Request , response : AbstractResponse ): String = {
78+ var requestDesc = " Request"
79+ var responseDesc = " Response"
80+ try {
81+ if (request == null ) {
82+ requestDesc += " null"
83+ } else {
84+ requestDesc += (" header: " + request.header)
85+ requestDesc += (" from service with principal: " +
86+ request.session.sanitizedUser +
87+ " IP address: " + request.session.clientAddress)
88+ }
89+ requestDesc += " | " // Separate the response description from the request description
90+
91+ if (response == null ) {
92+ responseDesc += " null"
93+ } else {
94+ responseDesc += (if (response.errorCounts == null || response.errorCounts.size == 0 ) {
95+ " with no error"
96+ } else {
97+ " with errors: " + response.errorCounts
98+ })
99+ }
100+ } catch {
101+ case e : Exception => return e.toString // If describing fails, return the exception message directly
102+ }
103+ requestDesc + responseDesc
104+ }
105+ }
0 commit comments