Add ssl listeners in channelInitializer #45

jiazhai · 2019-11-11T02:52:13Z

Add ssl listeners and sslCtxRefresher in channelInitializer

sijie · 2019-11-11T03:01:21Z

src/main/java/io/streamnative/kop/KafkaProtocolHandler.java

-        // TODO: support data register, when do https://github.com/streamnative/kop/issues/2
-        return "mock-data-for-kafka";
+        log.debug("Get configured listeners", kafkaConfig.getListeners());
+        return kafkaConfig.getListeners();


nit:

if (log.isDebugEnabled()) { log.debug("Get configured listeners", kafkaConfig.getListeners()); }

sijie · 2019-11-11T03:03:26Z

src/main/java/io/streamnative/kop/KafkaProtocolHandler.java

+            for (String listener: parts) {
+                if (listener.startsWith(PLAINTEXT_PREFIX)) {
+                    builder.put(
+                        new InetSocketAddress(brokerService.pulsar().getBindAddress(), getListenerPort(listener)),


shall we use the address in the listener as the bind address?

sijie · 2019-11-11T03:05:02Z

src/main/java/io/streamnative/kop/KafkaProtocolHandler.java

+                            true));
+                } else {
+                    log.error("KafkaProtocolHandler listeners {} not supported. supports {} and {}",
+                        listeners, PLAINTEXT_PREFIX, SSL_PREFIX);


Suggested change

listeners, PLAINTEXT_PREFIX, SSL_PREFIX);

listener, PLAINTEXT_PREFIX, SSL_PREFIX);

sijie · 2019-11-11T03:05:28Z

src/main/java/io/streamnative/kop/KafkaProtocolHandler.java

-            return initializerMap;
+                            true));
+                } else {
+                    log.error("KafkaProtocolHandler listeners {} not supported. supports {} and {}",


Suggested change

log.error("KafkaProtocolHandler listeners {} not supported. supports {} and {}",

log.error("Kafka listener {} is not supported. Only {} and {} are supported now.",

sijie · 2019-11-11T03:07:38Z

src/main/java/io/streamnative/kop/KafkaRequestHandler.java

@@ -119,11 +127,15 @@
    private final NamespaceName kafkaNamespace;
    private final ExecutorService executor;
    private final PulsarAdmin admin;
+    private final Boolean tlsEnabled;


why do we need tlsEnabled if we already support listeners?

In some method, it will need construct a Node(contains broker host and port), the port should be identified separately(9092 or 9093), so for this handler we need to know it is created with SSL enabled or not.

sijie · 2019-11-11T03:09:17Z

src/main/java/io/streamnative/kop/KafkaRequestHandler.java

+        if (!lookupResult.isPresent()) {
+            log.error("Can't find broker for topic {}", topic);
+            CompletableFuture<Optional<String>> future = new CompletableFuture<>();
+            future.completeExceptionally(new KeeperException.NoNodeException());


I don't think we should return a zookeeper exception here. We should return a Pulsar exception.

sijie · 2019-11-11T03:10:04Z

src/main/java/io/streamnative/kop/KafkaRequestHandler.java

+        } catch (Exception e) {
+            log.error("Failed to get URI from {} for topic {}", candidateBroker, topic);
+            CompletableFuture<Optional<String>> future = new CompletableFuture<>();
+            future.completeExceptionally(e);


Convert the exception into a broker exception?

sijie · 2019-11-11T03:24:53Z

src/main/java/io/streamnative/kop/KafkaRequestHandler.java

                } catch (Exception e) {
                    log.error("Caught error while find Broker for topic:{} ", topic, e);
-                    resultFuture.completeExceptionally(e);
+                    return null;


If we return null here, did we handle null in the caller?

add ssl listeners and sslCtxRefresher in channelInitializer

259cbeb

jiazhai requested a review from sijie November 11, 2019 02:52

jiazhai self-assigned this Nov 11, 2019

sijie suggested changes Nov 11, 2019

View reviewed changes

jiazhai added 6 commits November 12, 2019 11:38

fix typo prefix

f966443

pass test

fd09f4b

pass ut

a535f33

fix check

027ea77

change following sijie's comments

3332486

fix check

63bde88

sijie added area/protocol type/feature Indicates new functionality labels Nov 12, 2019

sijie added this to the 0.0.1 milestone Nov 12, 2019

sijie approved these changes Nov 13, 2019

View reviewed changes

sijie merged commit 3424abc into streamnative:master Nov 13, 2019

sijie deleted the enc_tls branch November 13, 2019 09:53

jiazhai changed the title ~~Add ssl listeners and sslCtxRefresher in channelInitializer~~ Add ssl listeners in channelInitializer Nov 13, 2019

jiazhai mentioned this pull request Jan 2, 2020

[FEATURE] Make SSL authorization work in Pulsar's way #56

Open

jiazhai modified the milestones: 0.0.1, 0.1.0 Feb 17, 2020

michaeljmarshall pushed a commit to michaeljmarshall/kop that referenced this pull request Jan 24, 2023

[tests] Allow SSH Access (streamnative#45)

806341c

This pull request was closed.

Provide feedback

Saved searches

Use saved searches to filter your results more quickly

Add ssl listeners in channelInitializer #45

Add ssl listeners in channelInitializer #45

jiazhai commented Nov 11, 2019

sijie Nov 11, 2019

sijie Nov 11, 2019

sijie Nov 11, 2019

sijie Nov 11, 2019

sijie Nov 11, 2019

jiazhai Nov 12, 2019

sijie Nov 11, 2019

sijie Nov 11, 2019

sijie Nov 11, 2019

	listeners, PLAINTEXT_PREFIX, SSL_PREFIX);
	listener, PLAINTEXT_PREFIX, SSL_PREFIX);

	log.error("KafkaProtocolHandler listeners {} not supported. supports {} and {}",
	log.error("Kafka listener {} is not supported. Only {} and {} are supported now.",

Add ssl listeners in channelInitializer #45

Add ssl listeners in channelInitializer #45

Conversation

jiazhai commented Nov 11, 2019

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