Skip to content

KAFKA-1683: persisting session information in Requests #155

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

Closed
gwenshap wants to merge 6 commits into from
Closed

KAFKA-1683: persisting session information in Requests #155

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
d97449d
persisting session information in Requests
gwenshap Aug 21, 2015
e55c50b
adding logging per Harsha's advice
gwenshap Aug 21, 2015
d4d4ed7
avoiding hardcoded ANONYMOUS string everywhere, per Parth's comment
gwenshap Aug 21, 2015
ba8a334
addressing suggestions by Ismael and Edward
gwenshap Aug 21, 2015
8d40258
added security protocol and principal to request logs
gwenshap Aug 24, 2015
48821bf
minor fix for logs following Jun's comments
gwenshap Aug 26, 2015
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
3 changes: 2 additions & 1 deletion clients/src/main/java/org/apache/kafka/common/network/SSLTransportLayer.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -33,6 +33,7 @@
import javax.net.ssl.SSLSession;
import javax.net.ssl.SSLPeerUnverifiedException;

import org.apache.kafka.common.security.auth.KafkaPrincipal;
import org.apache.kafka.common.utils.Utils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
Expand Down Expand Up @@ -600,7 +601,7 @@ public Principal peerPrincipal() throws IOException {
try {
return sslEngine.getSession().getPeerPrincipal();
} catch (SSLPeerUnverifiedException se) {

@harshach harshach Aug 21, 2015

Copy link
Copy Markdown

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

should we log a warning here

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

yeah, since we are caching the Principal in the authenticator, warning here makes sense.

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

btw. this is pretty weird - the session and the tests are on the server side, but since we are using the client network layer, you need to have log4j.logger.org.apache.kafka=WARN to see this warning.

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Just for my benefit, what would this warning indicate to a Kafka user? Can they do something about it?

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

This will indicate that even though they are using SSL, it is SSL without authentication (just encryption). It could be intentional, but could also be a misconfiguration (such as misplaced key files), hence warning.

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Understood, thanks.

throw new IOException(String.format("Unable to retrieve getPeerPrincipal due to %s", se));
return new KafkaPrincipal("ANONYMOUS");
}
}

Expand Down
7 changes: 5 additions & 2 deletions clients/src/main/java/org/apache/kafka/common/network/Selector.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -465,6 +465,7 @@ private void close(KafkaChannel channel) {
this.sensors.connectionClosed.record();
}


/**
* check if channel is ready
*/
Expand All @@ -475,9 +476,11 @@ public boolean isChannelReady(String id) {
}

/**
* Get the channel associated with this numeric id
* Get the channel associated with this connection
* Exposing this to allow SocketServer get the Principal from the channel when creating a request
* without making Selector know about Principals
*/
private KafkaChannel channelForId(String id) {
public KafkaChannel channelForId(String id) {
KafkaChannel channel = this.channels.get(id);
if (channel == null)
throw new IllegalStateException("Attempt to write to socket for which there is no open connection. Connection id " + id + " existing connections " + channels.keySet().toString());
Expand Down
5 changes: 4 additions & 1 deletion core/src/main/scala/kafka/network/RequestChannel.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package kafka.network

import java.nio.ByteBuffer
import java.security.Principal
import java.util.concurrent._

import com.yammer.metrics.core.Gauge
Expand All @@ -44,7 +45,9 @@ object RequestChannel extends Logging {
byteBuffer
}

case class Request(processor: Int, connectionId: String, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) {
case class Session(principal: Principal, host: String)

@eribeiro eribeiro Aug 21, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Session would have an id: Int to identify it, right? I mean, in the context of various "anonymous" principal connecting to the same host we need a way of uniquely identifying each session. Or would this id be the same as connectionId (i.e., no multiplexing multi-sessions on a single connection)? I don't know. wdyt?

update: I can see some methods/fields related to valid session state (valid, isValid, invalidate(), etc), but those can come in time, I think. :)

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Maybe the name is bad, but we didn't really plan on session IDs.
Sessions are always 1-1 with connections and those have the ID. The ID is srcHost:srcPort:destHost:destPort - which is guaranteed to be unique.

@eribeiro eribeiro Aug 21, 2015

Copy link
Copy Markdown
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, nice. :) 👍 Yeah... maybe ConnectionState or ConnectionInfo could be a better name. Just a suggestion.


case class Request(processor: Int, connectionId: String, session: Session = null, private var buffer: ByteBuffer, startTimeMs: Long, securityProtocol: SecurityProtocol) {

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Normally, one would use Option[Session] instead of a nullable one unless efficiency is of particular importance in that use-case. Is this one of those cases?

@gwenshap gwenshap Aug 21, 2015

Copy link
Copy Markdown
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nope :) Actually, since we always have a Session, I don't think Option is needed.
This was left from previous iteration.

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Oh, even better.

@volatile var requestDequeueTimeMs = -1L
@volatile var apiLocalCompleteTimeMs = -1L
@volatile var responseCompleteTimeMs = -1L
Expand Down
5 changes: 4 additions & 1 deletion core/src/main/scala/kafka/network/SocketServer.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -404,7 +404,10 @@ private[kafka] class Processor(val id: Int,
}
collection.JavaConversions.collectionAsScalaIterable(selector.completedReceives).foreach(receive => {
try {
val req = RequestChannel.Request(processor = id, connectionId = receive.source, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)

val channel = selector.channelForId(receive.source);
val session = RequestChannel.Session(channel.principal(), channel.socketDescription())

@ijuma ijuma Aug 21, 2015

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Nitpick: () is not needed unless it's a side-effecting method.

val req = RequestChannel.Request(processor = id, connectionId = receive.source, session = session, buffer = receive.payload, startTimeMs = time.milliseconds, securityProtocol = protocol)
requestChannel.sendRequest(req)
} catch {
case e @ (_: InvalidRequestException | _: SchemaException) => {
Expand Down
19 changes: 10 additions & 9 deletions core/src/test/scala/unit/kafka/network/SocketServerTest.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -17,21 +17,15 @@

package kafka.network;

import java.io._

import java.net._
import javax.net.ssl._
import java.io._
import java.nio.ByteBuffer
import java.util.Random

import kafka.api.ProducerRequest
import kafka.cluster.EndPoint
import kafka.common.TopicAndPartition
import kafka.message.ByteBufferMessageSet
import kafka.producer.SyncProducerConfig
import org.apache.kafka.common.metrics.Metrics
import org.apache.kafka.common.network.NetworkSend
import org.apache.kafka.common.protocol.SecurityProtocol
import org.apache.kafka.common.security.auth.KafkaPrincipal
import org.apache.kafka.common.utils.SystemTime
import org.junit.Assert._
import org.junit._
Expand All @@ -43,7 +37,6 @@ import java.nio.ByteBuffer
import kafka.common.TopicAndPartition
import kafka.message.ByteBufferMessageSet
import kafka.server.KafkaConfig
import java.nio.channels.SelectionKey
import kafka.utils.TestUtils

import scala.collection.Map
Expand Down Expand Up @@ -230,4 +223,12 @@ class SocketServerTest extends JUnitSuite {
assertEquals(serializedBytes.toSeq, receiveResponse(sslSocket).toSeq)
overrideServer.shutdown()
}

@Test
def testSessionPrincipal(): Unit = {
val socket = connect()
val bytes = new Array[Byte](40)
sendRequest(socket, 0, bytes)
assertEquals(new KafkaPrincipal("ANONYMOUS"),server.requestChannel.receiveRequest().session.principal)
}
}