[GitHub] [nifi] bejancsaba commented on a diff in pull request #6075: NIFI-9666 Implement base C2 client functionality

[GitBox <gi...@apache.org>]

bejancsaba commented on code in PR #6075:
URL: https://github.com/apache/nifi/pull/6075#discussion_r888827087


##########
c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Client.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * 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.c2.client.api;
+
+import java.util.Optional;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Defines interface methods used to implement a C2 Client. The controller can be application-specific but is used for such tasks as updating the flow.
+ */
+public interface C2Client {
+
+    Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) throws IOException;
+
+    ByteBuffer retrieveUpdateContent(String flowUpdateUrl);

Review Comment:
   There is no particular reason for ByteBuffer. I went with byte[] as that is coming natively from okhttp



##########
c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Client.java:
##########
@@ -0,0 +1,37 @@
+/*
+ * 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.c2.client.api;
+
+import java.util.Optional;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+
+/**
+ * Defines interface methods used to implement a C2 Client. The controller can be application-specific but is used for such tasks as updating the flow.
+ */
+public interface C2Client {
+
+    Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) throws IOException;

Review Comment:
   Added javadoc here and the neighbouring interfaces



##########
c2/c2-client-bundle/c2-client-base/src/main/java/org/apache/nifi/c2/serializer/C2JacksonSerializer.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.c2.serializer;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Optional;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2JacksonSerializer implements C2Serializer {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2JacksonSerializer.class);
+
+    private final ObjectMapper objectMapper;
+
+    public C2JacksonSerializer() {
+        objectMapper = new ObjectMapper();
+        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        objectMapper.enable(DeserializationFeature.READ_ENUMS_USING_TO_STRING);
+    }
+
+    @Override
+    public <T> Optional<String> serialize(T object) {
+        if (object == null) {
+            logger.trace("C2 Object was null. Nothing to serialize. Returning empty.");
+            return Optional.empty();
+        }
+
+        String contentString = null;
+        try {
+            contentString = objectMapper.writeValueAsString(object);
+            logger.trace("Serialized C2 Object: {}", contentString);
+        } catch (JsonProcessingException e) {
+            logger.error("Can't serialise C2 Object: ", e);
+        }
+
+        return Optional.ofNullable(contentString);
+    }
+
+    @Override
+    public <T> Optional<T> deserialize(String content, Class<T> valueType) {
+        if (content == null) {
+            logger.trace("Content for deserialization was null. Returning empty.");
+            return Optional.empty();
+        }
+
+        T responseObject = null;
+        try {
+            responseObject = objectMapper.readValue(content, valueType);
+        } catch (JsonProcessingException e) {
+            logger.error("Can't deserialize response object: ", e);

Review Comment:
   applied



##########
c2/c2-client-bundle/c2-client-http/pom.xml:
##########
@@ -0,0 +1,48 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>c2-client-bundle</artifactId>
+        <groupId>org.apache.nifi</groupId>
+        <version>1.17.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>c2-client-http</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-client-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-client-base</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>com.squareup.okhttp3</groupId>
+            <artifactId>okhttp</artifactId>
+            <version>${okhttp.version}</version>

Review Comment:
   removed, thanks



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);

Review Comment:
   Added HttpLoggingInterceptor so additional information with context can be logged for each request and response. Thanks for the idea it looks better now.



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);

Review Comment:
   Applied



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());
+            }
+
+            body = response.body();
+
+            if (body == null) {
+                logger.warn("No body returned when pulling a new configuration");
+                return null;
+            }
+
+            final ByteBuffer bodyByteBuffer = ByteBuffer.wrap(body.bytes());
+            return bodyByteBuffer;
+
+        } catch (Exception e) {
+            logger.warn("Hit an exception while trying to pull", e);
+            return null;
+        }
+    }
+
+    @Override
+    public void acknowledgeOperation(C2OperationAck operationAck) {
+        logger.info("Performing acknowledgement request to {} for operation {}", clientConfig.getC2AckUrl(), operationAck.getOperationId());
+        final ObjectMapper jacksonObjectMapper = new ObjectMapper();
+        jacksonObjectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        try {
+            final String operationAckBody = jacksonObjectMapper.writeValueAsString(operationAck);
+
+            final RequestBody requestBody = RequestBody.create(operationAckBody, MediaType.parse("application/json"));
+            final Request.Builder requestBuilder = new Request.Builder()
+                    .post(requestBody)
+                    .url(clientConfig.getC2AckUrl());
+            final Response heartbeatResponse = httpClientReference.get().newCall(requestBuilder.build()).execute();

Review Comment:
   Applied



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2HeartbeatFactory.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.c2.client.service;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.PersistentUuidGenerator;
+import org.apache.nifi.c2.client.service.model.RuntimeInfoWrapper;
+import org.apache.nifi.c2.protocol.api.AgentInfo;
+import org.apache.nifi.c2.protocol.api.AgentRepositories;
+import org.apache.nifi.c2.protocol.api.AgentStatus;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.DeviceInfo;
+import org.apache.nifi.c2.protocol.api.FlowInfo;
+import org.apache.nifi.c2.protocol.api.FlowQueueStatus;
+import org.apache.nifi.c2.protocol.api.NetworkInfo;
+import org.apache.nifi.c2.protocol.api.SystemInfo;
+import org.apache.nifi.c2.protocol.component.api.RuntimeManifest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HeartbeatFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HeartbeatFactory.class);
+
+    private static final String AGENT_IDENTIFIER_FILENAME = "agent-identifier";
+    private static final String DEVICE_IDENTIFIER_FILENAME = "device-identifier";
+
+    private final C2ClientConfig clientConfig;
+    private final FlowIdHolder flowIdHolder;
+
+    private String agentId;
+    private String deviceId;
+    private File confDirectory;
+
+    public C2HeartbeatFactory(C2ClientConfig clientConfig, FlowIdHolder flowIdHolder) {
+        this.clientConfig = clientConfig;
+        this.flowIdHolder = flowIdHolder;
+    }
+
+    public C2Heartbeat create(RuntimeInfoWrapper runtimeInfoWrapper) {
+        C2Heartbeat heartbeat = new C2Heartbeat();
+
+        heartbeat.setAgentInfo(getAgentInfo(runtimeInfoWrapper.getAgentRepositories(), runtimeInfoWrapper.getManifest()));
+        heartbeat.setDeviceInfo(generateDeviceInfo());
+        heartbeat.setFlowInfo(getFlowInfo(runtimeInfoWrapper.getQueueStatus()));
+        heartbeat.setCreated(System.currentTimeMillis());
+
+        return heartbeat;
+    }
+
+    //////
+    // FlowInfo
+    //////
+
+    private FlowInfo getFlowInfo(Map<String, FlowQueueStatus> queueStatus) {
+        FlowInfo flowInfo = new FlowInfo();
+        flowInfo.setQueues(queueStatus);
+        Optional.ofNullable(flowIdHolder.getFlowId()).ifPresent(flowInfo::setFlowId);
+        return flowInfo;
+    }
+
+    //////
+    // AgentInfo
+    //////
+
+    private AgentInfo getAgentInfo(AgentRepositories repos, RuntimeManifest manifest) {
+        AgentInfo agentInfo = new AgentInfo();
+        agentInfo.setAgentClass(clientConfig.getAgentClass());
+        agentInfo.setIdentifier(getAgentId());
+
+        AgentStatus agentStatus = new AgentStatus();
+        // TODO: Look into this, it doesn't look right
+        agentStatus.setUptime(System.currentTimeMillis());

Review Comment:
   Corrected



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/FlowIdHolder.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.c2.client.service;
+
+import static java.util.Collections.singletonList;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+import org.apache.nifi.util.file.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FlowIdHolder {
+    private static final Logger LOGGER = LoggerFactory.getLogger(FlowIdHolder.class);
+    private static final String FLOW_IDENTIFIER_FILENAME = "flow-identifier";
+
+    private volatile String flowId;
+    private final String configDirectoryName;
+
+    public FlowIdHolder(String configDirectoryName) {
+        this.configDirectoryName = configDirectoryName;
+        this.flowId = readFlowId();
+    }
+
+    public String getFlowId() {
+        return flowId;
+    }
+
+    public void setFlowId(String flowId) {
+        this.flowId = flowId;
+        persistFlowId(flowId);
+    }
+
+    private void persistFlowId(String flowId) {
+        File flowIdFile = new File(configDirectoryName, FLOW_IDENTIFIER_FILENAME);
+        try {
+            FileUtils.ensureDirectoryExistAndCanAccess(flowIdFile.getParentFile());
+            saveFlowId(flowIdFile, flowId);
+        } catch (IOException e) {
+            LOGGER.error("Failed to save flow information due to: {}", e.getMessage());
+        }
+    }
+
+    private void saveFlowId(File flowUpdateInfoFile, String flowId) {
+        try {
+            Files.write(flowUpdateInfoFile.toPath(), singletonList(flowId));
+        } catch (IOException e) {
+            LOGGER.error("Failed to persist flow data", e);

Review Comment:
   applied



##########
c2/c2-client-bundle/c2-client-api/src/main/java/org/apache/nifi/c2/client/api/C2Properties.java:
##########
@@ -0,0 +1,141 @@
+/*
+ * 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.c2.client.api;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.Properties;
+import java.util.concurrent.TimeUnit;
+
+public class C2Properties {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2Properties.class);
+
+    public static final String NIFI_PREFIX = "nifi.";
+
+    public static final String C2_PROPERTY_BASE = NIFI_PREFIX + "c2.";
+    public static final String C2_ENABLE_KEY = C2_PROPERTY_BASE + "enable";
+    public static final String C2_AGENT_PROTOCOL_KEY = C2_PROPERTY_BASE + "agent.protocol.class";
+    public static final String C2_REST_URL_KEY = C2_PROPERTY_BASE + "rest.url";
+    public static final String C2_REST_URL_ACK_KEY = C2_PROPERTY_BASE + "rest.url.ack";
+    public static final String C2_AGENT_HEARTBEAT_PERIOD_KEY = C2_PROPERTY_BASE + "agent.heartbeat.period";
+    public static final String C2_AGENT_CLASS_KEY = C2_PROPERTY_BASE + "agent.class";
+    public static final String C2_AGENT_IDENTIFIER_KEY = C2_PROPERTY_BASE + "agent.identifier";
+
+    public static final String C2_ROOT_CLASS_DEFINITIONS_KEY = C2_PROPERTY_BASE + "root.class.definitions";
+    public static final String C2_METRICS_NAME_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.name";
+    public static final String C2_METRICS_METRICS_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics";
+    public static final String C2_METRICS_METRICS_TYPED_METRICS_NAME_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics.typedmetrics.name";
+    public static final String C2_METRICS_METRICS_QUEUED_METRICS_NAME_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics.queuemetrics.name";
+    public static final String C2_METRICS_METRICS_QUEUE_METRICS_CLASSES_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics.queuemetrics.classes";
+    public static final String C2_METRICS_METRICS_TYPED_METRICS_CLASSES_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics.typedmetrics.classes";
+    public static final String C2_METRICS_METRICS_PROCESSOR_METRICS_NAME_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics.processorMetrics.name";
+    public static final String C2_METRICS_METRICS_PROCESSOR_METRICS_CLASSES_KEY = C2_ROOT_CLASS_DEFINITIONS_KEY + ".metrics.metrics.processorMetrics.classes";
+
+    /* REST Heartbeat Reporter Security Properties */
+    private static final String C2_REST_SECURITY_BASE_KEY = NIFI_PREFIX + "c2.security";
+    public static final String TRUSTSTORE_LOCATION_KEY = C2_REST_SECURITY_BASE_KEY + ".truststore.location";
+    public static final String TRUSTSTORE_PASSWORD_KEY = C2_REST_SECURITY_BASE_KEY + ".truststore.password";
+    public static final String TRUSTSTORE_TYPE_KEY = C2_REST_SECURITY_BASE_KEY + ".truststore.type";
+    public static final String KEYSTORE_LOCATION_KEY = C2_REST_SECURITY_BASE_KEY + ".keystore.location";
+    public static final String KEYSTORE_PASSWORD_KEY = C2_REST_SECURITY_BASE_KEY + ".keystore.password";
+    public static final String KEYSTORE_TYPE_KEY = C2_REST_SECURITY_BASE_KEY + ".keystore.type";
+    public static final String NEED_CLIENT_AUTH_KEY = C2_REST_SECURITY_BASE_KEY + ".need.client.auth";
+
+    // Defaults
+    // Heartbeat period of 1 second
+    public static final long C2_AGENT_DEFAULT_HEARTBEAT_PERIOD = TimeUnit.SECONDS.toMillis(1);

Review Comment:
   As you point out in one of your other comments this is a duplicated class (I suppose came back accidentally at a rebase) so I deleted it now.
   Reflecting on your comment: As you said there is already a lot of cleanup in the PR. We already identified other cleanup / refactor / hardening opportunities which we will cover after this PR gets the green light (didn't wanted to further extend this one). Would you be ok if we would look into this enum based approach there? I added to my list.



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandler.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.c2.client.service.operation;
+
+import static org.apache.commons.lang3.StringUtils.EMPTY;
+import static org.apache.nifi.c2.protocol.api.OperandType.CONFIGURATION;
+import static org.apache.nifi.c2.protocol.api.OperationType.UPDATE;
+
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.util.Optional;
+import java.util.function.Function;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.service.FlowIdHolder;
+import org.apache.nifi.c2.protocol.api.C2Operation;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.apache.nifi.c2.protocol.api.C2OperationState;
+import org.apache.nifi.c2.protocol.api.OperandType;
+import org.apache.nifi.c2.protocol.api.OperationType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class UpdateConfigurationOperationHandler implements C2OperationHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(UpdateConfigurationOperationHandler.class);
+
+    private static final String LOCATION = "location";
+
+    private final C2Client client;
+    private final Function<ByteBuffer, Boolean> updateFlow;
+    private final FlowIdHolder flowIdHolder;
+
+    public UpdateConfigurationOperationHandler(C2Client client, FlowIdHolder flowIdHolder, Function<ByteBuffer, Boolean> updateFlow) {
+        this.client = client;
+        this.updateFlow = updateFlow;
+        this.flowIdHolder = flowIdHolder;
+    }
+
+    @Override
+    public OperationType getOperationType() {
+        return UPDATE;
+    }
+
+    @Override
+    public OperandType getOperandType() {
+        return CONFIGURATION;
+    }
+
+    @Override
+    public C2OperationAck handle(C2Operation operation) {
+        String opIdentifier = Optional.ofNullable(operation.getIdentifier())
+            .orElse(EMPTY);
+        C2OperationAck operationAck = new C2OperationAck();
+        C2OperationState state = new C2OperationState();
+        operationAck.setOperationState(state);
+        operationAck.setOperationId(opIdentifier);
+
+        String updateLocation = Optional.ofNullable(operation.getArgs())
+            .map(map -> map.get(LOCATION))
+            .orElse(EMPTY);
+
+        String newFlowId = parseFlowId(updateLocation);
+        if (flowIdHolder.getFlowId() == null || !flowIdHolder.getFlowId().equals(newFlowId)) {
+            logger.info("Will perform flow update from {} for operation #{}. Previous flow id was {}, replacing with new id {}", updateLocation, opIdentifier,
+                flowIdHolder.getFlowId() == null ? "not set" : flowIdHolder.getFlowId(), newFlowId);
+        } else {
+            logger.info("Flow is current, no update is necessary...");

Review Comment:
   If you are ok with it I would leave it as is because flow update is a very rare activity while the outcome of the operation (whatever that may be) is really informative and descriptive in the logs.



##########
c2/c2-client-bundle/c2-client-base/pom.xml:
##########
@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>c2-client-bundle</artifactId>
+        <groupId>org.apache.nifi</groupId>
+        <version>1.17.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>c2-client-base</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-client-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-protocol-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-security-utils-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-security-utils</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-utils</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-databind</artifactId>
+        </dependency>
+        <dependency>
+            <groupId>com.fasterxml.jackson.core</groupId>
+            <artifactId>jackson-annotations</artifactId>
+            <version>2.9.0</version>

Review Comment:
   removed, thanks



##########
c2/c2-client-bundle/c2-client-base/pom.xml:
##########
@@ -0,0 +1,71 @@
+<?xml version="1.0" encoding="UTF-8"?>
+<!--
+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.
+ -->
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+
+    <parent>
+        <artifactId>c2-client-bundle</artifactId>
+        <groupId>org.apache.nifi</groupId>
+        <version>1.17.0-SNAPSHOT</version>
+    </parent>
+
+    <artifactId>c2-client-base</artifactId>
+    <packaging>jar</packaging>
+
+    <dependencies>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-client-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+            <scope>provided</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-protocol-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-security-utils-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>nifi-security-utils</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>

Review Comment:
   **nifi-utils:** was used to check directory exists and can be accessed. Moved those few lines of code here removed the dependency
   **nifi-security-utils:** was not needed at all so it was removed
   **nifi-security-utils-api:** the KeyStoreType enum was used. I followed the whole flow around C2NiFiClientService, httpClient and clientConfig and it looks like we don't need to rely on it. So remved and with that removed all nifi dependencies form here which is a good thing I suppose :)
   



##########
c2/c2-client-bundle/c2-client-base/src/main/java/org/apache/nifi/c2/serializer/C2JacksonSerializer.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.c2.serializer;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Optional;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2JacksonSerializer implements C2Serializer {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2JacksonSerializer.class);
+
+    private final ObjectMapper objectMapper;
+
+    public C2JacksonSerializer() {
+        objectMapper = new ObjectMapper();
+        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        objectMapper.enable(DeserializationFeature.READ_ENUMS_USING_TO_STRING);
+    }
+
+    @Override
+    public <T> Optional<String> serialize(T object) {
+        if (object == null) {
+            logger.trace("C2 Object was null. Nothing to serialize. Returning empty.");
+            return Optional.empty();
+        }
+
+        String contentString = null;
+        try {
+            contentString = objectMapper.writeValueAsString(object);
+            logger.trace("Serialized C2 Object: {}", contentString);
+        } catch (JsonProcessingException e) {
+            logger.error("Can't serialise C2 Object: ", e);

Review Comment:
   applied



##########
c2/c2-client-bundle/c2-client-base/src/main/java/org/apache/nifi/c2/serializer/C2JacksonSerializer.java:
##########
@@ -0,0 +1,75 @@
+/*
+ * 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.c2.serializer;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.DeserializationFeature;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.util.Optional;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2JacksonSerializer implements C2Serializer {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2JacksonSerializer.class);
+
+    private final ObjectMapper objectMapper;
+
+    public C2JacksonSerializer() {
+        objectMapper = new ObjectMapper();
+        objectMapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, false);
+        objectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        objectMapper.enable(DeserializationFeature.READ_ENUMS_USING_TO_STRING);
+    }
+
+    @Override
+    public <T> Optional<String> serialize(T object) {
+        if (object == null) {
+            logger.trace("C2 Object was null. Nothing to serialize. Returning empty.");
+            return Optional.empty();
+        }
+
+        String contentString = null;
+        try {
+            contentString = objectMapper.writeValueAsString(object);
+            logger.trace("Serialized C2 Object: {}", contentString);

Review Comment:
   Removed



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);

Review Comment:
   applied



##########
c2/c2-client-bundle/c2-client-base/src/main/java/org/apache/nifi/c2/client/PersistentUuidGenerator.java:
##########
@@ -0,0 +1,76 @@
+/*
+ * 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.c2.client;
+
+import org.apache.nifi.c2.client.api.IdGenerator;
+import org.apache.nifi.util.file.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.Arrays;
+import java.util.List;
+import java.util.UUID;
+
+public class PersistentUuidGenerator implements IdGenerator {
+
+    private static final Logger logger = LoggerFactory.getLogger(PersistentUuidGenerator.class);
+
+    private final File persistenceLocation;
+
+    public PersistentUuidGenerator(final File persistenceLocation) {
+        this.persistenceLocation = persistenceLocation;
+    }
+
+    @Override
+    public String generate() {
+        if (this.persistenceLocation.exists()) {
+            return readFile();
+        } else {
+            return makeFile();
+        }
+    }
+
+    private String readFile() {
+        try {
+            final List<String> fileLines = Files.readAllLines(persistenceLocation.toPath());
+            if (fileLines.size() != 1) {
+                throw new IllegalStateException(String.format("The file %s for the persisted identifier has the incorrect format.", persistenceLocation));
+            }
+            final String uuid = fileLines.get(0);
+            return uuid;
+        } catch (IOException e) {
+            throw new IllegalStateException(String.format("Could not read file %s for persisted identifier.", persistenceLocation), e);
+
+        }
+    }
+
+    private String makeFile() {
+        try {
+            final File parentDirectory = persistenceLocation.getParentFile();
+            FileUtils.ensureDirectoryExistAndCanAccess(parentDirectory);
+            final String uuid = UUID.randomUUID().toString();
+            Files.write(persistenceLocation.toPath(), Arrays.asList(uuid));
+            logger.info("Created identifier {} at {}.", uuid, persistenceLocation);

Review Comment:
   applied



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();

Review Comment:
   Good catch, updated.



##########
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-framework-core/src/main/java/org/apache/nifi/c2/C2NiFiProperties.java:
##########
@@ -0,0 +1,62 @@
+/*
+ * 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.c2;
+
+import java.util.concurrent.TimeUnit;
+
+public class C2NiFiProperties {

Review Comment:
   Actually we needed to keep this one and removed the other.



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();

Review Comment:
   Added property and configuration for setting call timeout.



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);

Review Comment:
   Applied



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);

Review Comment:
   Applied



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());

Review Comment:
   Applied the changed messaging. Regarding the extension of IOException, similarly to earlier this part of the code is inherited and we plan to do additional enhancements after this PR is merged. Would you be ok if I would look into it at that point? Added to my list.



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();

Review Comment:
   Thanks for the details, changed to exception throwing as proposed.



##########
minifi/pom.xml:
##########
@@ -449,10 +449,22 @@ limitations under the License.
                 <version>1.17.0-SNAPSHOT</version>
             </dependency>
 
+            <!-- C2 modules -->
+            <dependency>
+                <groupId>org.apache.nifi</groupId>
+                <artifactId>c2-client-api</artifactId>
+                <version>1.17.0-SNAPSHOT</version>
+            </dependency>
+            <dependency>
+                <groupId>org.apache.nifi</groupId>
+                <artifactId>c2-client-base</artifactId>
+                <version>1.17.0-SNAPSHOT</version>
+            </dependency>
+
             <dependency>
                 <groupId>com.squareup.okhttp3</groupId>
                 <artifactId>okhttp</artifactId>
-                <version>3.12.3</version>
+                <version>${okhttp.version}</version>

Review Comment:
   Removed



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());
+            }
+
+            body = response.body();
+
+            if (body == null) {
+                logger.warn("No body returned when pulling a new configuration");
+                return null;
+            }
+
+            final ByteBuffer bodyByteBuffer = ByteBuffer.wrap(body.bytes());
+            return bodyByteBuffer;
+
+        } catch (Exception e) {
+            logger.warn("Hit an exception while trying to pull", e);

Review Comment:
   Applied



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());
+            }
+
+            body = response.body();
+
+            if (body == null) {
+                logger.warn("No body returned when pulling a new configuration");
+                return null;
+            }
+
+            final ByteBuffer bodyByteBuffer = ByteBuffer.wrap(body.bytes());
+            return bodyByteBuffer;
+
+        } catch (Exception e) {
+            logger.warn("Hit an exception while trying to pull", e);
+            return null;
+        }
+    }
+
+    @Override
+    public void acknowledgeOperation(C2OperationAck operationAck) {
+        logger.info("Performing acknowledgement request to {} for operation {}", clientConfig.getC2AckUrl(), operationAck.getOperationId());
+        final ObjectMapper jacksonObjectMapper = new ObjectMapper();
+        jacksonObjectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);

Review Comment:
   Good catch, C2Serialiser was introduced just for this purpose and tried to use it everywhere it looks like I missed this spot. Updated.



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());
+            }
+
+            body = response.body();
+
+            if (body == null) {
+                logger.warn("No body returned when pulling a new configuration");
+                return null;
+            }
+
+            final ByteBuffer bodyByteBuffer = ByteBuffer.wrap(body.bytes());
+            return bodyByteBuffer;
+
+        } catch (Exception e) {
+            logger.warn("Hit an exception while trying to pull", e);
+            return null;
+        }
+    }
+
+    @Override
+    public void acknowledgeOperation(C2OperationAck operationAck) {
+        logger.info("Performing acknowledgement request to {} for operation {}", clientConfig.getC2AckUrl(), operationAck.getOperationId());
+        final ObjectMapper jacksonObjectMapper = new ObjectMapper();
+        jacksonObjectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        try {
+            final String operationAckBody = jacksonObjectMapper.writeValueAsString(operationAck);
+
+            final RequestBody requestBody = RequestBody.create(operationAckBody, MediaType.parse("application/json"));
+            final Request.Builder requestBuilder = new Request.Builder()
+                    .post(requestBody)
+                    .url(clientConfig.getC2AckUrl());
+            final Response heartbeatResponse = httpClientReference.get().newCall(requestBuilder.build()).execute();
+            if (!heartbeatResponse.isSuccessful()) {
+                logger.warn("Acknowledgement was not successful.");

Review Comment:
   Updated



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());
+            }
+
+            body = response.body();
+
+            if (body == null) {
+                logger.warn("No body returned when pulling a new configuration");
+                return null;
+            }
+
+            final ByteBuffer bodyByteBuffer = ByteBuffer.wrap(body.bytes());
+            return bodyByteBuffer;
+
+        } catch (Exception e) {
+            logger.warn("Hit an exception while trying to pull", e);
+            return null;
+        }
+    }
+
+    @Override
+    public void acknowledgeOperation(C2OperationAck operationAck) {
+        logger.info("Performing acknowledgement request to {} for operation {}", clientConfig.getC2AckUrl(), operationAck.getOperationId());
+        final ObjectMapper jacksonObjectMapper = new ObjectMapper();
+        jacksonObjectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        try {
+            final String operationAckBody = jacksonObjectMapper.writeValueAsString(operationAck);
+
+            final RequestBody requestBody = RequestBody.create(operationAckBody, MediaType.parse("application/json"));

Review Comment:
   We actually already have a constant for this called MEDIA_TYPE_APPLICATION_JSON but it looks like this place was missed. Corrected.



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2HeartbeatFactory.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.c2.client.service;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.PersistentUuidGenerator;
+import org.apache.nifi.c2.client.service.model.RuntimeInfoWrapper;
+import org.apache.nifi.c2.protocol.api.AgentInfo;
+import org.apache.nifi.c2.protocol.api.AgentRepositories;
+import org.apache.nifi.c2.protocol.api.AgentStatus;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.DeviceInfo;
+import org.apache.nifi.c2.protocol.api.FlowInfo;
+import org.apache.nifi.c2.protocol.api.FlowQueueStatus;
+import org.apache.nifi.c2.protocol.api.NetworkInfo;
+import org.apache.nifi.c2.protocol.api.SystemInfo;
+import org.apache.nifi.c2.protocol.component.api.RuntimeManifest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HeartbeatFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HeartbeatFactory.class);
+
+    private static final String AGENT_IDENTIFIER_FILENAME = "agent-identifier";
+    private static final String DEVICE_IDENTIFIER_FILENAME = "device-identifier";
+
+    private final C2ClientConfig clientConfig;
+    private final FlowIdHolder flowIdHolder;
+
+    private String agentId;
+    private String deviceId;
+    private File confDirectory;
+
+    public C2HeartbeatFactory(C2ClientConfig clientConfig, FlowIdHolder flowIdHolder) {
+        this.clientConfig = clientConfig;
+        this.flowIdHolder = flowIdHolder;
+    }
+
+    public C2Heartbeat create(RuntimeInfoWrapper runtimeInfoWrapper) {
+        C2Heartbeat heartbeat = new C2Heartbeat();
+
+        heartbeat.setAgentInfo(getAgentInfo(runtimeInfoWrapper.getAgentRepositories(), runtimeInfoWrapper.getManifest()));
+        heartbeat.setDeviceInfo(generateDeviceInfo());
+        heartbeat.setFlowInfo(getFlowInfo(runtimeInfoWrapper.getQueueStatus()));
+        heartbeat.setCreated(System.currentTimeMillis());
+
+        return heartbeat;
+    }
+
+    //////
+    // FlowInfo
+    //////

Review Comment:
   Removed



##########
c2/c2-client-bundle/c2-client-http/src/main/java/org/apache/nifi/c2/client/http/C2HttpClient.java:
##########
@@ -0,0 +1,249 @@
+/*
+ * 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.c2.client.http;
+
+import com.fasterxml.jackson.annotation.JsonInclude;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import java.io.FileInputStream;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.security.KeyStore;
+import java.security.NoSuchAlgorithmException;
+import java.util.Optional;
+import java.util.concurrent.atomic.AtomicReference;
+import javax.net.ssl.KeyManagerFactory;
+import javax.net.ssl.SSLContext;
+import javax.net.ssl.SSLSocketFactory;
+import javax.net.ssl.TrustManager;
+import javax.net.ssl.TrustManagerFactory;
+import javax.net.ssl.X509TrustManager;
+import okhttp3.MediaType;
+import okhttp3.OkHttpClient;
+import okhttp3.Request;
+import okhttp3.RequestBody;
+import okhttp3.Response;
+import okhttp3.ResponseBody;
+import org.apache.commons.lang3.StringUtils;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.api.C2Serializer;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HttpClient implements C2Client {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HttpClient.class);
+    private static final MediaType MEDIA_TYPE_APPLICATION_JSON = MediaType.parse("application/json");
+
+    private final AtomicReference<OkHttpClient> httpClientReference = new AtomicReference<>();
+    private final C2ClientConfig clientConfig;
+    private final C2Serializer serializer;
+
+    public C2HttpClient(C2ClientConfig clientConfig, C2Serializer serializer) {
+        super();
+        this.clientConfig = clientConfig;
+        this.serializer = serializer;
+        final OkHttpClient.Builder okHttpClientBuilder = new OkHttpClient.Builder();
+
+        // Set whether to follow redirects
+        okHttpClientBuilder.followRedirects(true);
+
+        // check if the ssl path is set and add the factory if so
+        if (StringUtils.isNotBlank(clientConfig.getKeystoreFilename())) {
+            try {
+                setSslSocketFactory(okHttpClientBuilder);
+            } catch (Exception e) {
+                throw new IllegalStateException(e);
+            }
+        }
+
+        httpClientReference.set(okHttpClientBuilder.build());
+    }
+
+    @Override
+    public Optional<C2HeartbeatResponse> publishHeartbeat(C2Heartbeat heartbeat) {
+        return serializer.serialize(heartbeat).flatMap(this::sendHeartbeat);
+    }
+
+    private Optional<C2HeartbeatResponse> sendHeartbeat(String heartbeat) {
+        logger.debug("Sending heartbeat to {}", clientConfig.getC2Url());
+
+        Optional<C2HeartbeatResponse> c2HeartbeatResponse = Optional.empty();
+        Request request = new Request.Builder()
+            .post(RequestBody.create(heartbeat, MEDIA_TYPE_APPLICATION_JSON))
+            .url(clientConfig.getC2Url())
+            .build();
+
+        try {
+            Response heartbeatResponse = httpClientReference.get().newCall(request).execute();
+            c2HeartbeatResponse = getResponseBody(heartbeatResponse).flatMap(response -> serializer.deserialize(response, C2HeartbeatResponse.class));
+        } catch (IOException ce) {
+            logger.error("Error while sending heartbeat", ce);
+        }
+
+        return c2HeartbeatResponse;
+    }
+
+    private Optional<String> getResponseBody(Response response) {
+        String responseBody = null;
+
+        try {
+            responseBody = response.body().string();
+            logger.debug("Received response (Status={}) {}", response.code(), responseBody);
+        } catch (IOException e) {
+            logger.error("Could not get response body: ", e);
+        }
+
+        return Optional.ofNullable(responseBody);
+    }
+
+    private void setSslSocketFactory(OkHttpClient.Builder okHttpClientBuilder) throws Exception {
+        final String keystoreLocation = clientConfig.getKeystoreFilename();
+        final String keystoreType = clientConfig.getKeystoreType().getType();
+        final String keystorePass = clientConfig.getKeystorePass();
+
+        assertKeystorePropertiesSet(keystoreLocation, keystorePass, keystoreType);
+
+        // prepare the keystore
+        final KeyStore keyStore = KeyStore.getInstance(keystoreType);
+
+        try (FileInputStream keyStoreStream = new FileInputStream(keystoreLocation)) {
+            keyStore.load(keyStoreStream, keystorePass.toCharArray());
+        }
+
+        final KeyManagerFactory keyManagerFactory = KeyManagerFactory.getInstance(KeyManagerFactory.getDefaultAlgorithm());
+        keyManagerFactory.init(keyStore, keystorePass.toCharArray());
+
+        // load truststore
+        final String truststoreLocation = clientConfig.getTruststoreFilename();
+        final String truststorePass = clientConfig.getTruststorePass();
+        final String truststoreType = clientConfig.getTruststoreType().getType();
+        assertTruststorePropertiesSet(truststoreLocation, truststorePass, truststoreType);
+
+        KeyStore truststore = KeyStore.getInstance(truststoreType);
+        final TrustManagerFactory trustManagerFactory = TrustManagerFactory.getInstance("X509");
+        truststore.load(new FileInputStream(truststoreLocation), truststorePass.toCharArray());
+        trustManagerFactory.init(truststore);
+
+        final X509TrustManager x509TrustManager;
+        TrustManager[] trustManagers = trustManagerFactory.getTrustManagers();
+        if (trustManagers[0] != null) {
+            x509TrustManager = (X509TrustManager) trustManagers[0];
+        } else {
+            throw new IllegalStateException("List of trust managers is null");
+        }
+
+        SSLContext tempSslContext;
+        try {
+            tempSslContext = SSLContext.getInstance("TLS");
+        } catch (NoSuchAlgorithmException e) {
+            logger.warn("Unable to use 'TLS' for the PullHttpChangeIngestor due to NoSuchAlgorithmException. Will attempt to use the default algorithm.", e);
+            tempSslContext = SSLContext.getDefault();
+        }
+
+        final SSLContext sslContext = tempSslContext;
+        sslContext.init(keyManagerFactory.getKeyManagers(), trustManagerFactory.getTrustManagers(), null);
+
+        final SSLSocketFactory sslSocketFactory = sslContext.getSocketFactory();
+        okHttpClientBuilder.sslSocketFactory(sslSocketFactory, x509TrustManager);
+    }
+
+    private void assertKeystorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException(clientConfig.getKeystoreFilename() + " is null or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's keystore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    private void assertTruststorePropertiesSet(String location, String password, String type) {
+        if (location == null || location.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is not set or is empty");
+        }
+
+        if (password == null || password.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its password is not (or is empty). If the location is set, the password must also be.");
+        }
+
+        if (type == null || type.isEmpty()) {
+            throw new IllegalArgumentException("The client's truststore filename is set but its type is not (or is empty). If the location is set, the type must also be.");
+        }
+    }
+
+    @Override
+    public ByteBuffer retrieveUpdateContent(String flowUpdateUrl) {
+        final Request.Builder requestBuilder = new Request.Builder()
+                .get()
+                .url(flowUpdateUrl);
+        final Request request = requestBuilder.build();
+
+        ResponseBody body;
+        try (final Response response = httpClientReference.get().newCall(request).execute()) {
+            logger.debug("Response received: {}", response);
+
+            int code = response.code();
+            if (code >= 400) {
+                throw new IOException("Got response code " + code + " while trying to pull configuration: " + response.body().string());
+            }
+
+            body = response.body();
+
+            if (body == null) {
+                logger.warn("No body returned when pulling a new configuration");
+                return null;
+            }
+
+            final ByteBuffer bodyByteBuffer = ByteBuffer.wrap(body.bytes());
+            return bodyByteBuffer;
+
+        } catch (Exception e) {
+            logger.warn("Hit an exception while trying to pull", e);
+            return null;
+        }
+    }
+
+    @Override
+    public void acknowledgeOperation(C2OperationAck operationAck) {
+        logger.info("Performing acknowledgement request to {} for operation {}", clientConfig.getC2AckUrl(), operationAck.getOperationId());
+        final ObjectMapper jacksonObjectMapper = new ObjectMapper();
+        jacksonObjectMapper.setSerializationInclusion(JsonInclude.Include.NON_NULL);
+        try {
+            final String operationAckBody = jacksonObjectMapper.writeValueAsString(operationAck);
+
+            final RequestBody requestBody = RequestBody.create(operationAckBody, MediaType.parse("application/json"));
+            final Request.Builder requestBuilder = new Request.Builder()
+                    .post(requestBody)
+                    .url(clientConfig.getC2AckUrl());
+            final Response heartbeatResponse = httpClientReference.get().newCall(requestBuilder.build()).execute();
+            if (!heartbeatResponse.isSuccessful()) {
+                logger.warn("Acknowledgement was not successful.");
+            }
+            logger.trace("Status on acknowledgement was {}", heartbeatResponse.code());

Review Comment:
   I removed this particular log (but added status elsewhere) also with the okhttp logging interceptor this should be covered.



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2ClientService.java:
##########
@@ -0,0 +1,78 @@
+/*
+ * 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.c2.client.service;
+
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.Arrays;
+import java.util.List;
+import java.util.function.Function;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.service.model.RuntimeInfoWrapper;
+import org.apache.nifi.c2.client.service.operation.C2OperationService;
+import org.apache.nifi.c2.client.service.operation.UpdateConfigurationOperationHandler;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.C2HeartbeatResponse;
+import org.apache.nifi.c2.protocol.api.C2Operation;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2ClientService {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2ClientService.class);
+
+    private final C2Client client;
+    private final C2HeartbeatFactory c2HeartbeatFactory;
+    private final C2OperationService operationService;
+    private final UpdateConfigurationOperationHandler updateConfigurationOperationHandler;
+
+    public C2ClientService(C2Client client, C2HeartbeatFactory c2HeartbeatFactory, FlowIdHolder flowIdHolder, Function<ByteBuffer, Boolean> updateFlow) {
+        this.client = client;
+        this.c2HeartbeatFactory = c2HeartbeatFactory;
+        this.updateConfigurationOperationHandler = new UpdateConfigurationOperationHandler(client, flowIdHolder, updateFlow);
+        this.operationService = new C2OperationService(Arrays.asList(updateConfigurationOperationHandler));
+    }
+
+    public void sendHeartbeat(RuntimeInfoWrapper runtimeInfoWrapper) {
+        try {
+            // TODO exception handling for all the C2 Client interactions (IOExceptions, logger.error vs logger.warn, etc.)
+            C2Heartbeat c2Heartbeat = c2HeartbeatFactory.create(runtimeInfoWrapper);
+            client.publishHeartbeat(c2Heartbeat).ifPresent(this::processResponse);
+        } catch (IOException ioe) {
+            // TODO

Review Comment:
   These TODOs were left here from the "original approach" after the rewrites they had no point I removed them (for example as the function returns optional with the latest implementation there was no exception thrown so the try/catch became obsolete. I removed it. Thanks.



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2HeartbeatFactory.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.c2.client.service;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.PersistentUuidGenerator;
+import org.apache.nifi.c2.client.service.model.RuntimeInfoWrapper;
+import org.apache.nifi.c2.protocol.api.AgentInfo;
+import org.apache.nifi.c2.protocol.api.AgentRepositories;
+import org.apache.nifi.c2.protocol.api.AgentStatus;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.DeviceInfo;
+import org.apache.nifi.c2.protocol.api.FlowInfo;
+import org.apache.nifi.c2.protocol.api.FlowQueueStatus;
+import org.apache.nifi.c2.protocol.api.NetworkInfo;
+import org.apache.nifi.c2.protocol.api.SystemInfo;
+import org.apache.nifi.c2.protocol.component.api.RuntimeManifest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HeartbeatFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HeartbeatFactory.class);
+
+    private static final String AGENT_IDENTIFIER_FILENAME = "agent-identifier";
+    private static final String DEVICE_IDENTIFIER_FILENAME = "device-identifier";
+
+    private final C2ClientConfig clientConfig;
+    private final FlowIdHolder flowIdHolder;
+
+    private String agentId;
+    private String deviceId;
+    private File confDirectory;
+
+    public C2HeartbeatFactory(C2ClientConfig clientConfig, FlowIdHolder flowIdHolder) {
+        this.clientConfig = clientConfig;
+        this.flowIdHolder = flowIdHolder;
+    }
+
+    public C2Heartbeat create(RuntimeInfoWrapper runtimeInfoWrapper) {
+        C2Heartbeat heartbeat = new C2Heartbeat();
+
+        heartbeat.setAgentInfo(getAgentInfo(runtimeInfoWrapper.getAgentRepositories(), runtimeInfoWrapper.getManifest()));
+        heartbeat.setDeviceInfo(generateDeviceInfo());
+        heartbeat.setFlowInfo(getFlowInfo(runtimeInfoWrapper.getQueueStatus()));
+        heartbeat.setCreated(System.currentTimeMillis());
+
+        return heartbeat;
+    }
+
+    //////
+    // FlowInfo
+    //////
+
+    private FlowInfo getFlowInfo(Map<String, FlowQueueStatus> queueStatus) {
+        FlowInfo flowInfo = new FlowInfo();
+        flowInfo.setQueues(queueStatus);
+        Optional.ofNullable(flowIdHolder.getFlowId()).ifPresent(flowInfo::setFlowId);
+        return flowInfo;
+    }
+
+    //////
+    // AgentInfo
+    //////
+
+    private AgentInfo getAgentInfo(AgentRepositories repos, RuntimeManifest manifest) {
+        AgentInfo agentInfo = new AgentInfo();
+        agentInfo.setAgentClass(clientConfig.getAgentClass());
+        agentInfo.setIdentifier(getAgentId());
+
+        AgentStatus agentStatus = new AgentStatus();
+        // TODO: Look into this, it doesn't look right
+        agentStatus.setUptime(System.currentTimeMillis());
+        agentStatus.setRepositories(repos);
+
+        agentInfo.setStatus(agentStatus);
+        agentInfo.setAgentManifest(manifest);
+
+        return agentInfo;
+    }
+
+    private String getAgentId() {
+        if (agentId == null) {
+            String rawAgentId = clientConfig.getAgentIdentifier();
+            if (isNotBlank(rawAgentId)) {
+                agentId = rawAgentId.trim();
+            } else {
+                File idFile = new File(getConfDirectory(), AGENT_IDENTIFIER_FILENAME);
+                agentId = new PersistentUuidGenerator(idFile).generate();
+            }
+        }
+
+        return agentId;
+    }
+
+    //////
+    // DeviceInfo
+    //////
+
+    private DeviceInfo generateDeviceInfo() {
+        // Populate DeviceInfo
+        final DeviceInfo deviceInfo = new DeviceInfo();
+        deviceInfo.setNetworkInfo(generateNetworkInfo());
+        deviceInfo.setIdentifier(getDeviceIdentifier(deviceInfo.getNetworkInfo()));
+        deviceInfo.setSystemInfo(generateSystemInfo());
+        return deviceInfo;
+    }
+
+    private NetworkInfo generateNetworkInfo() {
+        NetworkInfo networkInfo = new NetworkInfo();
+        try {
+            // Determine all interfaces
+            final Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
+
+            final Set<NetworkInterface> operationIfaces = new HashSet<>();
+
+            // Determine eligible interfaces
+            while (networkInterfaces.hasMoreElements()) {
+                final NetworkInterface networkInterface = networkInterfaces.nextElement();
+                if (!networkInterface.isLoopback() && networkInterface.isUp()) {
+                    operationIfaces.add(networkInterface);
+                }
+            }
+            logger.trace("Have {} interfaces with names {}", operationIfaces.size(),
+                operationIfaces.stream()
+                    .map(NetworkInterface::getName)
+                    .collect(Collectors.toSet())
+            );
+
+            if (!operationIfaces.isEmpty()) {
+                if (operationIfaces.size() > 1) {
+                    logger.debug("Instance has multiple interfaces.  Generated information may be non-deterministic.");
+                }
+
+                NetworkInterface iface = operationIfaces.iterator().next();
+                final StringBuilder macSb = new StringBuilder();
+                byte[] hardwareAddress = iface.getHardwareAddress();
+                if (hardwareAddress != null) {
+                    for (int i = 0; i < hardwareAddress.length; i++) {
+                        macSb.append(String.format("%02X", hardwareAddress[i]));
+                    }
+                }
+                final String macString = macSb.toString();
+                Enumeration<InetAddress> inetAddresses = iface.getInetAddresses();
+                while (inetAddresses.hasMoreElements()) {
+                    InetAddress inetAddress = inetAddresses.nextElement();
+                    String hostAddress = inetAddress.getHostAddress();
+                    String hostName = inetAddress.getHostName();
+                    byte[] address = inetAddress.getAddress();
+                    String canonicalHostName = inetAddress.getCanonicalHostName();
+
+                    networkInfo.setDeviceId(iface.getName());
+                    networkInfo.setHostname(hostName);
+                    networkInfo.setIpAddress(hostAddress);
+                }
+            }
+        } catch (
+            Exception e) {
+            logger.error("Had exception determining network information", e);

Review Comment:
   Applied



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/FlowIdHolder.java:
##########
@@ -0,0 +1,85 @@
+/*
+ * 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.c2.client.service;
+
+import static java.util.Collections.singletonList;
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.file.Files;
+import java.util.List;
+import org.apache.nifi.util.file.FileUtils;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class FlowIdHolder {
+    private static final Logger LOGGER = LoggerFactory.getLogger(FlowIdHolder.class);
+    private static final String FLOW_IDENTIFIER_FILENAME = "flow-identifier";
+
+    private volatile String flowId;
+    private final String configDirectoryName;
+
+    public FlowIdHolder(String configDirectoryName) {
+        this.configDirectoryName = configDirectoryName;
+        this.flowId = readFlowId();
+    }
+
+    public String getFlowId() {
+        return flowId;
+    }
+
+    public void setFlowId(String flowId) {
+        this.flowId = flowId;
+        persistFlowId(flowId);
+    }
+
+    private void persistFlowId(String flowId) {
+        File flowIdFile = new File(configDirectoryName, FLOW_IDENTIFIER_FILENAME);
+        try {
+            FileUtils.ensureDirectoryExistAndCanAccess(flowIdFile.getParentFile());
+            saveFlowId(flowIdFile, flowId);
+        } catch (IOException e) {
+            LOGGER.error("Failed to save flow information due to: {}", e.getMessage());

Review Comment:
   applied



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/C2OperationHandler.java:
##########
@@ -0,0 +1,31 @@
+/*
+ * 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.c2.client.service.operation;
+
+import org.apache.nifi.c2.protocol.api.C2Operation;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.apache.nifi.c2.protocol.api.OperandType;
+import org.apache.nifi.c2.protocol.api.OperationType;
+
+public interface C2OperationHandler {
+
+    OperationType getOperationType();
+
+    OperandType getOperandType();
+
+    C2OperationAck handle(C2Operation operation);

Review Comment:
   Added



##########
minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/util/ProcessUtils.java:
##########
@@ -0,0 +1,131 @@
+/*
+ * 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.minifi.bootstrap.util;
+
+import java.io.BufferedReader;
+import java.io.IOException;
+import java.io.InputStream;
+import java.io.InputStreamReader;
+import java.io.Reader;
+import java.util.ArrayList;
+import java.util.List;
+import java.util.concurrent.TimeUnit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class ProcessUtils {

Review Comment:
   @ferencerdei could you please take a look?



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/operation/UpdateConfigurationOperationHandler.java:
##########
@@ -0,0 +1,113 @@
+/*
+ * 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.c2.client.service.operation;
+
+import static org.apache.commons.lang3.StringUtils.EMPTY;
+import static org.apache.nifi.c2.protocol.api.OperandType.CONFIGURATION;
+import static org.apache.nifi.c2.protocol.api.OperationType.UPDATE;
+
+import java.net.URI;
+import java.nio.ByteBuffer;
+import java.util.Optional;
+import java.util.function.Function;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.service.FlowIdHolder;
+import org.apache.nifi.c2.protocol.api.C2Operation;
+import org.apache.nifi.c2.protocol.api.C2OperationAck;
+import org.apache.nifi.c2.protocol.api.C2OperationState;
+import org.apache.nifi.c2.protocol.api.OperandType;
+import org.apache.nifi.c2.protocol.api.OperationType;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class UpdateConfigurationOperationHandler implements C2OperationHandler {
+
+    private static final Logger logger = LoggerFactory.getLogger(UpdateConfigurationOperationHandler.class);
+
+    private static final String LOCATION = "location";
+
+    private final C2Client client;
+    private final Function<ByteBuffer, Boolean> updateFlow;
+    private final FlowIdHolder flowIdHolder;
+
+    public UpdateConfigurationOperationHandler(C2Client client, FlowIdHolder flowIdHolder, Function<ByteBuffer, Boolean> updateFlow) {
+        this.client = client;
+        this.updateFlow = updateFlow;
+        this.flowIdHolder = flowIdHolder;
+    }
+
+    @Override
+    public OperationType getOperationType() {
+        return UPDATE;
+    }
+
+    @Override
+    public OperandType getOperandType() {
+        return CONFIGURATION;
+    }
+
+    @Override
+    public C2OperationAck handle(C2Operation operation) {
+        String opIdentifier = Optional.ofNullable(operation.getIdentifier())
+            .orElse(EMPTY);
+        C2OperationAck operationAck = new C2OperationAck();
+        C2OperationState state = new C2OperationState();
+        operationAck.setOperationState(state);
+        operationAck.setOperationId(opIdentifier);
+
+        String updateLocation = Optional.ofNullable(operation.getArgs())
+            .map(map -> map.get(LOCATION))
+            .orElse(EMPTY);
+
+        String newFlowId = parseFlowId(updateLocation);
+        if (flowIdHolder.getFlowId() == null || !flowIdHolder.getFlowId().equals(newFlowId)) {
+            logger.info("Will perform flow update from {} for operation #{}. Previous flow id was {}, replacing with new id {}", updateLocation, opIdentifier,
+                flowIdHolder.getFlowId() == null ? "not set" : flowIdHolder.getFlowId(), newFlowId);
+        } else {
+            logger.info("Flow is current, no update is necessary...");
+        }
+
+        flowIdHolder.setFlowId(newFlowId);
+        ByteBuffer updateContent = client.retrieveUpdateContent(updateLocation);
+        if (updateContent != null) {
+            if (updateFlow.apply(updateContent)) {
+                state.setState(C2OperationState.OperationState.FULLY_APPLIED);
+                logger.debug("Update configuration applied for operation #{}.", opIdentifier);
+            } else {
+                state.setState(C2OperationState.OperationState.NOT_APPLIED);
+                logger.error("Update resulted in error for operation #{}.", opIdentifier);
+            }
+        } else {
+            state.setState(C2OperationState.OperationState.NOT_APPLIED);
+            logger.error("Update content retrieval resulted in empty content so flow update was omitted for operation #{}.", opIdentifier);
+        }
+
+        return operationAck;
+    }
+
+    private String parseFlowId(String flowUpdateUrl) {
+        try {
+            URI flowUri = new URI(flowUpdateUrl);
+            String flowUriPath = flowUri.getPath();
+            String[] split = flowUriPath.split("/");
+            String flowId = split[4];

Review Comment:
   There is a change currently under development on the server side which will affect this so we will make it more robust when that will be ready. Till then I added the check as proposed.



##########
minifi/minifi-bootstrap/src/main/java/org/apache/nifi/minifi/bootstrap/command/EnvRunner.java:
##########
@@ -0,0 +1,107 @@
+/*
+ * 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.minifi.bootstrap.command;
+
+import static org.apache.nifi.minifi.bootstrap.RunMiNiFi.CMD_LOGGER;
+import static org.apache.nifi.minifi.bootstrap.RunMiNiFi.DEFAULT_LOGGER;
+import static org.apache.nifi.minifi.bootstrap.Status.ERROR;
+import static org.apache.nifi.minifi.bootstrap.Status.MINIFI_NOT_RUNNING;
+import static org.apache.nifi.minifi.bootstrap.Status.OK;
+
+import java.lang.reflect.Method;
+import java.util.Map.Entry;
+import java.util.Properties;
+import org.apache.nifi.minifi.bootstrap.MiNiFiParameters;
+import org.apache.nifi.minifi.bootstrap.MiNiFiStatus;
+import org.apache.nifi.minifi.bootstrap.service.MiNiFiStatusProvider;
+
+public class EnvRunner implements CommandRunner {
+    private final MiNiFiParameters miNiFiParameters;
+    private final MiNiFiStatusProvider miNiFiStatusProvider;
+
+    public EnvRunner(MiNiFiParameters miNiFiParameters, MiNiFiStatusProvider miNiFiStatusProvider) {
+        this.miNiFiParameters = miNiFiParameters;
+        this.miNiFiStatusProvider = miNiFiStatusProvider;
+    }
+
+    /**
+     * Returns information about the MiNiFi's virtual machine.
+     * @param args the input arguments
+     * @return status code
+     */
+    @Override
+    public int runCommand(String[] args) {
+        return env();
+    }
+
+    private int env() {
+        MiNiFiStatus status = miNiFiStatusProvider.getStatus(miNiFiParameters.getMiNiFiPort(), miNiFiParameters.getMinifiPid());
+        if (status.getPid() == null) {
+            CMD_LOGGER.error("Apache MiNiFi is not running");
+            return MINIFI_NOT_RUNNING.getStatusCode();
+        }
+
+        Class<?> virtualMachineClass;
+        try {
+            virtualMachineClass = Class.forName("com.sun.tools.attach.VirtualMachine");

Review Comment:
   Talked with @ferencerdei we will validate. 



##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2HeartbeatFactory.java:
##########
@@ -0,0 +1,248 @@
+/*
+ * 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.c2.client.service;
+
+import static org.apache.commons.lang3.StringUtils.isNotBlank;
+
+import java.io.File;
+import java.lang.management.ManagementFactory;
+import java.lang.management.OperatingSystemMXBean;
+import java.net.InetAddress;
+import java.net.NetworkInterface;
+import java.util.Enumeration;
+import java.util.HashSet;
+import java.util.Map;
+import java.util.Optional;
+import java.util.Set;
+import java.util.stream.Collectors;
+import org.apache.nifi.c2.client.C2ClientConfig;
+import org.apache.nifi.c2.client.PersistentUuidGenerator;
+import org.apache.nifi.c2.client.service.model.RuntimeInfoWrapper;
+import org.apache.nifi.c2.protocol.api.AgentInfo;
+import org.apache.nifi.c2.protocol.api.AgentRepositories;
+import org.apache.nifi.c2.protocol.api.AgentStatus;
+import org.apache.nifi.c2.protocol.api.C2Heartbeat;
+import org.apache.nifi.c2.protocol.api.DeviceInfo;
+import org.apache.nifi.c2.protocol.api.FlowInfo;
+import org.apache.nifi.c2.protocol.api.FlowQueueStatus;
+import org.apache.nifi.c2.protocol.api.NetworkInfo;
+import org.apache.nifi.c2.protocol.api.SystemInfo;
+import org.apache.nifi.c2.protocol.component.api.RuntimeManifest;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2HeartbeatFactory {
+
+    private static final Logger logger = LoggerFactory.getLogger(C2HeartbeatFactory.class);
+
+    private static final String AGENT_IDENTIFIER_FILENAME = "agent-identifier";
+    private static final String DEVICE_IDENTIFIER_FILENAME = "device-identifier";
+
+    private final C2ClientConfig clientConfig;
+    private final FlowIdHolder flowIdHolder;
+
+    private String agentId;
+    private String deviceId;
+    private File confDirectory;
+
+    public C2HeartbeatFactory(C2ClientConfig clientConfig, FlowIdHolder flowIdHolder) {
+        this.clientConfig = clientConfig;
+        this.flowIdHolder = flowIdHolder;
+    }
+
+    public C2Heartbeat create(RuntimeInfoWrapper runtimeInfoWrapper) {
+        C2Heartbeat heartbeat = new C2Heartbeat();
+
+        heartbeat.setAgentInfo(getAgentInfo(runtimeInfoWrapper.getAgentRepositories(), runtimeInfoWrapper.getManifest()));
+        heartbeat.setDeviceInfo(generateDeviceInfo());
+        heartbeat.setFlowInfo(getFlowInfo(runtimeInfoWrapper.getQueueStatus()));
+        heartbeat.setCreated(System.currentTimeMillis());
+
+        return heartbeat;
+    }
+
+    //////
+    // FlowInfo
+    //////
+
+    private FlowInfo getFlowInfo(Map<String, FlowQueueStatus> queueStatus) {
+        FlowInfo flowInfo = new FlowInfo();
+        flowInfo.setQueues(queueStatus);
+        Optional.ofNullable(flowIdHolder.getFlowId()).ifPresent(flowInfo::setFlowId);
+        return flowInfo;
+    }
+
+    //////
+    // AgentInfo
+    //////
+
+    private AgentInfo getAgentInfo(AgentRepositories repos, RuntimeManifest manifest) {
+        AgentInfo agentInfo = new AgentInfo();
+        agentInfo.setAgentClass(clientConfig.getAgentClass());
+        agentInfo.setIdentifier(getAgentId());
+
+        AgentStatus agentStatus = new AgentStatus();
+        // TODO: Look into this, it doesn't look right
+        agentStatus.setUptime(System.currentTimeMillis());
+        agentStatus.setRepositories(repos);
+
+        agentInfo.setStatus(agentStatus);
+        agentInfo.setAgentManifest(manifest);
+
+        return agentInfo;
+    }
+
+    private String getAgentId() {
+        if (agentId == null) {
+            String rawAgentId = clientConfig.getAgentIdentifier();
+            if (isNotBlank(rawAgentId)) {
+                agentId = rawAgentId.trim();
+            } else {
+                File idFile = new File(getConfDirectory(), AGENT_IDENTIFIER_FILENAME);
+                agentId = new PersistentUuidGenerator(idFile).generate();
+            }
+        }
+
+        return agentId;
+    }
+
+    //////
+    // DeviceInfo
+    //////
+
+    private DeviceInfo generateDeviceInfo() {
+        // Populate DeviceInfo
+        final DeviceInfo deviceInfo = new DeviceInfo();
+        deviceInfo.setNetworkInfo(generateNetworkInfo());
+        deviceInfo.setIdentifier(getDeviceIdentifier(deviceInfo.getNetworkInfo()));
+        deviceInfo.setSystemInfo(generateSystemInfo());
+        return deviceInfo;
+    }
+
+    private NetworkInfo generateNetworkInfo() {
+        NetworkInfo networkInfo = new NetworkInfo();
+        try {
+            // Determine all interfaces
+            final Enumeration<NetworkInterface> networkInterfaces = NetworkInterface.getNetworkInterfaces();
+
+            final Set<NetworkInterface> operationIfaces = new HashSet<>();
+
+            // Determine eligible interfaces
+            while (networkInterfaces.hasMoreElements()) {
+                final NetworkInterface networkInterface = networkInterfaces.nextElement();
+                if (!networkInterface.isLoopback() && networkInterface.isUp()) {
+                    operationIfaces.add(networkInterface);
+                }
+            }
+            logger.trace("Have {} interfaces with names {}", operationIfaces.size(),
+                operationIfaces.stream()
+                    .map(NetworkInterface::getName)
+                    .collect(Collectors.toSet())
+            );
+
+            if (!operationIfaces.isEmpty()) {
+                if (operationIfaces.size() > 1) {
+                    logger.debug("Instance has multiple interfaces.  Generated information may be non-deterministic.");
+                }
+
+                NetworkInterface iface = operationIfaces.iterator().next();
+                final StringBuilder macSb = new StringBuilder();
+                byte[] hardwareAddress = iface.getHardwareAddress();
+                if (hardwareAddress != null) {
+                    for (int i = 0; i < hardwareAddress.length; i++) {
+                        macSb.append(String.format("%02X", hardwareAddress[i]));
+                    }
+                }
+                final String macString = macSb.toString();
+                Enumeration<InetAddress> inetAddresses = iface.getInetAddresses();
+                while (inetAddresses.hasMoreElements()) {
+                    InetAddress inetAddress = inetAddresses.nextElement();
+                    String hostAddress = inetAddress.getHostAddress();
+                    String hostName = inetAddress.getHostName();
+                    byte[] address = inetAddress.getAddress();
+                    String canonicalHostName = inetAddress.getCanonicalHostName();
+
+                    networkInfo.setDeviceId(iface.getName());
+                    networkInfo.setHostname(hostName);
+                    networkInfo.setIpAddress(hostAddress);
+                }
+            }
+        } catch (
+            Exception e) {
+            logger.error("Had exception determining network information", e);
+        }
+        return networkInfo;
+    }
+
+    private String getDeviceIdentifier(NetworkInfo networkInfo) {
+        if (deviceId == null) {
+            if (networkInfo.getDeviceId() != null) {
+                try {
+                    final NetworkInterface netInterface = NetworkInterface.getByName(networkInfo.getDeviceId());
+                    byte[] hardwareAddress = netInterface.getHardwareAddress();
+                    final StringBuilder macBuilder = new StringBuilder();
+                    if (hardwareAddress != null) {
+                        for (byte address : hardwareAddress) {
+                            macBuilder.append(String.format("%02X", address));

Review Comment:
   Removed the code duplication as its result was not used.



##########
nifi-server-api/pom.xml:
##########
@@ -42,5 +42,10 @@
             <version>1.17.0-SNAPSHOT</version>
             <scope>provided</scope>
         </dependency>
+        <dependency>
+            <groupId>org.apache.nifi</groupId>
+            <artifactId>c2-client-api</artifactId>
+            <version>1.17.0-SNAPSHOT</version>
+        </dependency>

Review Comment:
   I think this is the result of a semi successful rebase, you are right it is not needed, removed.



##########
minifi/minifi-nar-bundles/minifi-framework-bundle/minifi-server-nar/pom.xml:
##########
@@ -43,6 +43,12 @@
             <artifactId>nifi-headless-server</artifactId>
             <version>1.17.0-SNAPSHOT</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.commons</groupId>
+            <artifactId>commons-lang3</artifactId>
+            <version>${org.apache.commons.lang3.version}</version>

Review Comment:
   Removed



-- 
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: issues-unsubscribe@nifi.apache.org

For queries about this service, please contact Infrastructure at:
users@infra.apache.org