briansolo1985 commented on code in PR #8738:
URL: https://github.com/apache/nifi/pull/8738#discussion_r1590813383
##########
nifi-framework-api/src/main/java/org/apache/nifi/bootstrap/BootstrapCommunicator.java:
##########
@@ -25,11 +25,13 @@ public interface BootstrapCommunicator {
/**
* Sends a command with specific arguments to the bootstrap process
+ *
* @param command the command to send
- * @param args the args to send
+ * @param args the args to send
+ * @return whether the command send was successful or not
* @throws IOException exception in case of communication issue
*/
- void sendCommand(String command, String... args) throws IOException;
+ boolean sendCommand(String command, String... args) throws IOException;
Review Comment:
Fixed
##########
c2/c2-client-bundle/c2-client-service/src/main/java/org/apache/nifi/c2/client/service/C2OperationManager.java:
##########
@@ -0,0 +1,204 @@
+/*
+ * 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.Optional.of;
+import static java.util.Optional.ofNullable;
+import static java.util.function.Predicate.not;
+import static
org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.FULLY_APPLIED;
+import static
org.apache.nifi.c2.protocol.api.C2OperationState.OperationState.NOT_APPLIED;
+
+import java.util.LinkedList;
+import java.util.List;
+import java.util.Optional;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.service.operation.C2OperationHandler;
+import org.apache.nifi.c2.client.service.operation.C2OperationHandlerProvider;
+import org.apache.nifi.c2.client.service.operation.C2OperationRestartHandler;
+import org.apache.nifi.c2.client.service.operation.OperationQueue;
+import org.apache.nifi.c2.client.service.operation.OperationQueueDAO;
+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.C2OperationState.OperationState;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+public class C2OperationManager implements Runnable {
+
+ private static final Logger LOGGER =
LoggerFactory.getLogger(C2OperationManager.class);
+
+ private final C2Client client;
+ private final C2OperationHandlerProvider c2OperationHandlerProvider;
+ private final ReentrantLock heartbeatLock;
+ private final OperationQueueDAO operationQueueDAO;
+ private final C2OperationRestartHandler c2OperationRestartHandler;
+ private final BlockingQueue<C2Operation> c2Operations;
+
+ public C2OperationManager(C2Client client, C2OperationHandlerProvider
c2OperationHandlerProvider, ReentrantLock heartbeatLock,
+ OperationQueueDAO operationQueueDAO,
C2OperationRestartHandler c2OperationRestartHandler) {
+ this.client = client;
+ this.c2OperationHandlerProvider = c2OperationHandlerProvider;
+ this.heartbeatLock = heartbeatLock;
+ this.operationQueueDAO = operationQueueDAO;
+ this.c2OperationRestartHandler = c2OperationRestartHandler;
+ this.c2Operations = new LinkedBlockingQueue<>();
+ }
+
+ public void add(C2Operation c2Operation) {
+ try {
+ c2Operations.put(c2Operation);
+ } catch (InterruptedException e) {
+ LOGGER.warn("Thread was interrupted", e);
+ }
+ }
+
+ @Override
+ public void run() {
+ processRestartState();
+
+ while (true) {
+ C2Operation operation;
+ try {
+ operation = c2Operations.take();
+ } catch (InterruptedException e) {
+ LOGGER.warn("Thread was interrupted", e);
+ return;
+ }
+
+ LOGGER.debug("Processing operation {}", operation);
+ C2OperationHandler operationHandler =
c2OperationHandlerProvider.getHandlerForOperation(operation).orElse(null);
+ if (operationHandler == null) {
+ LOGGER.debug("No handler is present for for C2 Operation {}",
operation);
Review Comment:
Removed
##########
c2/c2-client-bundle/c2-client-service/src/test/java/org/apache/nifi/c2/client/service/C2HeartbeatManagerTest.java:
##########
@@ -0,0 +1,140 @@
+/*
+ * 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.Optional.empty;
+import static java.util.Optional.ofNullable;
+import static org.mockito.ArgumentMatchers.any;
+import static org.mockito.Mockito.mock;
+import static org.mockito.Mockito.never;
+import static org.mockito.Mockito.times;
+import static org.mockito.Mockito.verify;
+import static org.mockito.Mockito.when;
+
+import java.util.List;
+import java.util.concurrent.locks.ReentrantLock;
+import org.apache.nifi.c2.client.api.C2Client;
+import org.apache.nifi.c2.client.service.model.RuntimeInfoWrapper;
+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.junit.jupiter.api.Test;
+import org.junit.jupiter.api.extension.ExtendWith;
+import org.mockito.InjectMocks;
+import org.mockito.Mock;
+import org.mockito.junit.jupiter.MockitoExtension;
+
+@ExtendWith(MockitoExtension.class)
+public class C2HeartbeatManagerTest {
+
+ @Mock
+ private C2Client mockC2Client;
+
+ @Mock
+ private C2HeartbeatFactory mockC2HeartbeatFactory;
+
+ @Mock
+ private ReentrantLock mockHeartbeatLock;
+
+ @Mock
+ private RuntimeInfoWrapper mockRuntimeInfoWrapper;
+
+ @Mock
+ private C2OperationManager mockC2OperationManager;
+
+ @InjectMocks
+ private C2HeartbeatManager testHeartbeatManager;
+
+ @Test
+ void shouldSkipSendingHeartbeatIfHeartbeatLockIsAcquired() {
+ when(mockHeartbeatLock.tryLock()).thenReturn(false);
+
+ testHeartbeatManager.run();
+
+ verify(mockC2HeartbeatFactory, never()).create(any());
+ verify(mockC2Client, never()).publishHeartbeat(any());
+ verify(mockC2OperationManager, never()).add(any());
+ verify(mockHeartbeatLock, never()).unlock();
+ }
+
+ @Test
+ void shouldSendHeartbeatAndProcessEmptyResponse() {
+ when(mockHeartbeatLock.tryLock()).thenReturn(true);
+ C2Heartbeat mockC2Heartbeat = mock(C2Heartbeat.class);
+
when(mockC2HeartbeatFactory.create(mockRuntimeInfoWrapper)).thenReturn(mockC2Heartbeat);
+
when(mockC2Client.publishHeartbeat(mockC2Heartbeat)).thenReturn(empty());
+
+ testHeartbeatManager.run();
+
+ verify(mockC2HeartbeatFactory,
times(1)).create(mockRuntimeInfoWrapper);
+ verify(mockC2Client, times(1)).publishHeartbeat(mockC2Heartbeat);
+ verify(mockC2OperationManager, never()).add(any());
+ verify(mockHeartbeatLock, times(1)).unlock();
+
+ }
+
+ @Test
+ void shouldSendHeartbeatAndProcessResponseWithNoOperation() {
+ when(mockHeartbeatLock.tryLock()).thenReturn(true);
+ C2Heartbeat mockC2Heartbeat = mock(C2Heartbeat.class);
+
when(mockC2HeartbeatFactory.create(mockRuntimeInfoWrapper)).thenReturn(mockC2Heartbeat);
+ C2HeartbeatResponse mockC2HeartbeatResponse =
mock(C2HeartbeatResponse.class);
+
when(mockC2HeartbeatResponse.getRequestedOperations()).thenReturn(List.of());
+
when(mockC2Client.publishHeartbeat(mockC2Heartbeat)).thenReturn(ofNullable(mockC2HeartbeatResponse));
Review Comment:
Yes, fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
