exceptionfactory commented on a change in pull request #5311: URL: https://github.com/apache/nifi/pull/5311#discussion_r717085052
########## File path: nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/DistributedCacheClient.java ########## @@ -0,0 +1,72 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.distributed.cache.client; + +import io.netty.channel.Channel; +import io.netty.channel.pool.ChannelPool; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.distributed.cache.client.adapter.InboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.OutboundAdapter; +import org.apache.nifi.remote.VersionNegotiatorFactory; + +import java.io.IOException; + +/** + * Encapsulate operations which may be performed using a {@link DistributedSetCacheClientService} or a + * {@link DistributedMapCacheClientService}. + */ +public class DistributedCacheClient { + + /** + * The pool of network connections used to service client requests. + */ + private final ChannelPool channelPool; + + /** + * Constructor. + * + * @param context the NiFi configuration to be applied to the channel pool + * @param factory creator of object used to broker the version of the distributed cache protocol with the service + */ + protected DistributedCacheClient(final ConfigurationContext context, final VersionNegotiatorFactory factory) { Review comment: This constructor will need to change to pass individual property values. ########## File path: nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/NettyDistributedMapCacheClient.java ########## @@ -0,0 +1,309 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.distributed.cache.client; + +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.distributed.cache.client.adapter.AtomicCacheEntryInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.BooleanInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.LongInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.MapInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.MapValuesInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.OutboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.SetInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.ValueInboundAdapter; +import org.apache.nifi.distributed.cache.client.adapter.VoidInboundAdapter; +import org.apache.nifi.distributed.cache.operations.MapOperation; +import org.apache.nifi.distributed.cache.protocol.ProtocolVersion; +import org.apache.nifi.remote.VersionNegotiatorFactory; + +import java.io.IOException; +import java.util.Collection; +import java.util.Map; +import java.util.Set; + +/** + * The implementation of the {@link DistributedMapCacheClient} using the netty library to provide the remote + * communication services. + */ +public class NettyDistributedMapCacheClient extends DistributedCacheClient { + + /** + * Constructor. + * + * @param context the NiFi configuration to be applied to the channel pool + * @param factory creator of object used to broker the version of the distributed cache protocol with the service + */ + public NettyDistributedMapCacheClient(final ConfigurationContext context, final VersionNegotiatorFactory factory) { Review comment: This constructor will need to change to pass individual property values. ########## File path: nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-client-service/src/main/java/org/apache/nifi/distributed/cache/client/CacheClientChannelPoolFactory.java ########## @@ -0,0 +1,89 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.distributed.cache.client; + +import io.netty.bootstrap.Bootstrap; +import io.netty.channel.Channel; +import io.netty.channel.ChannelOption; +import io.netty.channel.EventLoopGroup; +import io.netty.channel.nio.NioEventLoopGroup; +import io.netty.channel.pool.ChannelHealthChecker; +import io.netty.channel.pool.ChannelPool; +import io.netty.channel.pool.ChannelPoolHandler; +import io.netty.channel.pool.FixedChannelPool; +import io.netty.channel.socket.nio.NioSocketChannel; +import org.apache.nifi.components.PropertyValue; +import org.apache.nifi.controller.ConfigurationContext; +import org.apache.nifi.event.transport.netty.channel.pool.InitializingChannelPoolHandler; +import org.apache.nifi.remote.VersionNegotiatorFactory; +import org.apache.nifi.ssl.SSLContextService; + +import javax.net.ssl.SSLContext; +import java.util.concurrent.TimeUnit; + +/** + * Factory for construction of new {@link ChannelPool}, used by distributed cache clients to invoke service + * methods. Cache clients include the NiFi services {@link DistributedSetCacheClientService} + * and {@link DistributedMapCacheClientService}. + */ +public class CacheClientChannelPoolFactory { + + private static final int MAX_PENDING_ACQUIRES = 1024; + + private int maxConnections = Runtime.getRuntime().availableProcessors() * 2; + + /** + * Set Maximum Connections for Channel Pool + * + * @param maxConnections Maximum Number of connections defaults to available processors multiplied by 2 + */ + public void setMaxConnections(final int maxConnections) { + this.maxConnections = maxConnections; + } + + /** + * Instantiate a new netty pool of channels to be used for distributed cache communications + * + * @param context the NiFi configuration to be applied to the channel pool + * @param factory creator of object used to broker the version of the distributed cache protocol with the service + * @return a channel pool object from which {@link Channel} objects may be obtained + */ + public ChannelPool createChannelPool(final ConfigurationContext context, final VersionNegotiatorFactory factory) { + final String hostname = context.getProperty(DistributedSetCacheClientService.HOSTNAME).getValue(); + final int port = context.getProperty(DistributedSetCacheClientService.PORT).asInteger(); + final PropertyValue timeoutMillis = context.getProperty(DistributedSetCacheClientService.COMMUNICATIONS_TIMEOUT); + final SSLContextService sslContextService = context.getProperty( + DistributedSetCacheClientService.SSL_CONTEXT_SERVICE).asControllerService(SSLContextService.class); + final SSLContext sslContext = (sslContextService == null) ? null : sslContextService.createContext(); Review comment: These `PropertyDescriptor` references point to `DistributedSetCacheClientService`, but this class is also shared with `DistributedMapCacheClientService`. For this reason, the method should be refactored to take individual parameters, and dependent services will need to perform property resolution at a higher level. ```suggestion public ChannelPool createChannelPool(final String hostname, final int port, final int timeoutMillis, final SSLContext sslContext, final VersionNegotiatorFactory factory) { ``` ########## File path: nifi-nar-bundles/nifi-standard-services/nifi-distributed-cache-services-bundle/nifi-distributed-cache-server/src/test/java/org/apache/nifi/distributed/cache/server/map/DistributedMapCacheTlsTest.java ########## @@ -0,0 +1,142 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one or more + * contributor license agreements. See the NOTICE file distributed with + * this work for additional information regarding copyright ownership. + * The ASF licenses this file to You under the Apache License, Version 2.0 + * (the "License"); you may not use this file except in compliance with + * the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.apache.nifi.distributed.cache.server.map; + +import org.apache.commons.lang3.SerializationException; +import org.apache.nifi.distributed.cache.client.Deserializer; +import org.apache.nifi.distributed.cache.client.Serializer; +import org.apache.nifi.distributed.cache.client.exception.DeserializationException; +import org.apache.nifi.distributed.cache.client.DistributedMapCacheClientService; +import org.apache.nifi.processor.Processor; +import org.apache.nifi.security.util.KeyStoreUtils; +import org.apache.nifi.security.util.SslContextFactory; +import org.apache.nifi.security.util.TlsConfiguration; +import org.apache.nifi.ssl.SSLContextService; +import org.apache.nifi.util.TestRunner; +import org.apache.nifi.util.TestRunners; +import org.junit.AfterClass; +import org.junit.BeforeClass; +import org.junit.Test; +import org.mockito.Mockito; + +import javax.net.ssl.SSLContext; +import java.io.File; +import java.io.IOException; +import java.io.OutputStream; +import java.nio.charset.StandardCharsets; +import java.security.GeneralSecurityException; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertFalse; +import static org.junit.Assert.assertTrue; + +/** + * Verify basic functionality of {@link DistributedMapCacheClientService}, in the context of a TLS authenticated + * socket session. + * <p> + * This test instantiates both the server and client {@link org.apache.nifi.controller.ControllerService} objects + * implementing the distributed cache protocol. It assumes that the default distributed cache port (4557) + * is available. + */ +public class DistributedMapCacheTlsTest { + + private static TestRunner runner = null; + private static SSLContextService sslContextService = null; + private static DistributedMapCacheServer server = null; + private static DistributedMapCacheClientService client = null; + private static final Serializer<String> serializer = new StringSerializer(); + private static final Deserializer<String> deserializer = new StringDeserializer(); + + @BeforeClass + public static void beforeClass() throws Exception { + final String port = DistributedMapCacheServer.PORT.getDefaultValue(); + runner = TestRunners.newTestRunner(Mockito.mock(Processor.class)); + sslContextService = createSslContextService(); + runner.addControllerService(sslContextService.getIdentifier(), sslContextService); + runner.enableControllerService(sslContextService); + + server = new DistributedMapCacheServer(); + runner.addControllerService(server.getClass().getName(), server); + runner.setProperty(server, DistributedMapCacheServer.PORT, port); + runner.setProperty(server, DistributedMapCacheServer.SSL_CONTEXT_SERVICE, sslContextService.getIdentifier()); + runner.enableControllerService(server); + + client = new DistributedMapCacheClientService(); + runner.addControllerService(client.getClass().getName(), client); + runner.setProperty(client, DistributedMapCacheClientService.HOSTNAME, "localhost"); + runner.setProperty(client, DistributedMapCacheClientService.PORT, port); + runner.setProperty(client, DistributedMapCacheClientService.SSL_CONTEXT_SERVICE, sslContextService.getIdentifier()); + runner.enableControllerService(client); + } + + @AfterClass + public static void afterClass() { + runner.disableControllerService(client); + runner.removeControllerService(client); + + runner.disableControllerService(server); + runner.removeControllerService(server); + + runner.disableControllerService(sslContextService); + runner.removeControllerService(sslContextService); + } + + @Test + public void testMapPut() throws IOException { + final String key = "keyPut"; + final String value = "valuePut"; + assertFalse(client.containsKey(key, serializer)); + client.put(key, value, serializer, serializer); + assertTrue(client.containsKey(key, serializer)); + assertEquals(value, client.get(key, serializer, deserializer)); + assertTrue(client.remove(key, serializer)); + assertFalse(client.containsKey(key, serializer)); + } + + /** + * Create a fresh {@link SSLContext} in order to test mutual TLS authentication aspect of the + * distributed cache protocol. + * + * @return a NiFi {@link SSLContextService}, to be used to secure the distributed cache comms + * @throws GeneralSecurityException on SSLContext generation failure + * @throws IOException on failure to serialize the SSLContext + */ + private static SSLContextService createSslContextService() throws GeneralSecurityException, IOException { + final TlsConfiguration tlsConfiguration = KeyStoreUtils.createTlsConfigAndNewKeystoreTruststore(); + new File(tlsConfiguration.getKeystorePath()).deleteOnExit(); + new File(tlsConfiguration.getTruststorePath()).deleteOnExit(); Review comment: After rebasing on the current main branch, this will need to change to use `new TemporaryKeyStoreBuilder().build()`: ```suggestion final TlsConfiguration tlsConfiguration = new TemporaryKeyStoreBuilder().build(); ``` -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected]
