@Test public void testReceiveRaw() throws Exception { setupMock(SiteToSiteTransportProtocol.RAW, TransferDirection.RECEIVE); setupMockProcessSession(); final String peerUrl = "nifi: final PeerDescription peerDescription = new PeerDescription("node1.example.com", 9090, true); try (final SocketChannel socketChannel = SocketChannel.open()) { final CommunicationsSession commsSession = new SocketChannelCommunicationsSession(socketChannel); commsSession.setUserDn("nifi.node1.example.com"); final Peer peer = new Peer(peerDescription, commsSession, peerUrl, REMOTE_CLUSTER_URL); doReturn(peer).when(transaction).getCommunicant(); final String sourceFlowFileUuid = "flowfile-uuid"; final Map<String, String> attributes = new HashMap<>(); attributes.put(CoreAttributes.UUID.key(), sourceFlowFileUuid); final byte[] dataPacketContents = "DataPacket Contents".getBytes(); final ByteArrayInputStream dataPacketInputStream = new ByteArrayInputStream(dataPacketContents); final DataPacket dataPacket = new StandardDataPacket(attributes, dataPacketInputStream, dataPacketContents.length); doReturn(dataPacket).doReturn(null).when(this.transaction).receive(); port.onTrigger(processContext, processSession); final List<ProvenanceEventRecord> provenanceEvents = sessionState.getProvenanceEvents(); assertEquals(1, provenanceEvents.size()); final ProvenanceEventRecord provenanceEvent = provenanceEvents.get(0); assertEquals(ProvenanceEventType.RECEIVE, provenanceEvent.getEventType()); assertEquals(peerUrl + "/" + sourceFlowFileUuid, provenanceEvent.getTransitUri()); assertEquals("Remote DN=nifi.node1.example.com", provenanceEvent.getDetails()); processSession.assertAllFlowFilesTransferred(Relationship.ANONYMOUS); final List<MockFlowFile> flowFiles = processSession.getFlowFilesForRelationship(Relationship.ANONYMOUS); assertEquals(1, flowFiles.size()); final MockFlowFile flowFile = flowFiles.get(0); flowFile.assertAttributeEquals(SiteToSiteAttributes.S2S_HOST.key(), peer.getHost()); flowFile.assertAttributeEquals(SiteToSiteAttributes.S2S_ADDRESS.key(), peer.getHost() + ":" + peer.getPort()); } }
@Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
            final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
        final NiFiProperties nifiProperties); }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
            final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
        final NiFiProperties nifiProperties); @Override boolean isTargetRunning(); void setTargetRunning(final boolean targetRunning); @Override boolean isTriggerWhenEmpty(); @Override Resource getResource(); @Override Authorizable getParentAuthorizable(); @Override void shutdown(); @Override void onSchedulingStart(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); @Override String getYieldPeriod(); @Override boolean getTargetExists(); @Override boolean isValid(); @Override Collection<ValidationResult> getValidationErrors(); @Override void verifyCanStart(); @Override void setUseCompression(final boolean useCompression); @Override boolean isUseCompression(); @Override String toString(); @Override RemoteProcessGroup getRemoteProcessGroup(); @Override TransferDirection getTransferDirection(); void setTargetExists(final boolean exists); @Override void removeConnection(final Connection connection); @Override SchedulingStrategy getSchedulingStrategy(); @Override boolean isSideEffectFree(); @Override String getComponentType(); }
StandardRemoteGroupPort extends RemoteGroupPort { @Override public void onTrigger(final ProcessContext context, final ProcessSession session) { if (!remoteGroup.isTransmitting()) { logger.debug("{} {} is not transmitting; will not send/receive", this, remoteGroup); return; } if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT && session.getQueueSize().getObjectCount() == 0) { logger.debug("{} No data to send", this); return; } final String url = getRemoteProcessGroup().getTargetUri(); final FlowFile firstFlowFile; if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { firstFlowFile = session.get(); if (firstFlowFile == null) { return; } } else { firstFlowFile = null; } final SiteToSiteClient client = getSiteToSiteClient(); final Transaction transaction; try { transaction = client.createTransaction(transferDirection); } catch (final PortNotRunningException e) { context.yield(); this.targetRunning.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port is not in a valid state", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final UnknownPortException e) { context.yield(); this.targetExists.set(false); final String message = String.format("%s failed to communicate with %s because the remote instance indicates that the port no longer exists", this, url); logger.error(message); session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } catch (final IOException e) { final String message = String.format("%s failed to communicate with %s due to %s", this, url, e.toString()); logger.error(message); if (logger.isDebugEnabled()) { logger.error("", e); } session.rollback(); remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); return; } if (transaction == null) { logger.debug("{} Unable to create transaction to communicate with; all peers must be penalized, so yielding context", this); session.rollback(); context.yield(); return; } try { if (getConnectableType() == ConnectableType.REMOTE_INPUT_PORT) { transferFlowFiles(transaction, context, session, firstFlowFile); } else { final int numReceived = receiveFlowFiles(transaction, context, session); if (numReceived == 0) { context.yield(); } } session.commit(); } catch (final Throwable t) { final String message = String.format("%s failed to communicate with remote NiFi instance due to %s", this, t.toString()); logger.error("{} failed to communicate with remote NiFi instance due to {}", this, t.toString()); if (logger.isDebugEnabled()) { logger.error("", t); } remoteGroup.getEventReporter().reportEvent(Severity.ERROR, CATEGORY, message); transaction.error(); session.rollback(); } } StandardRemoteGroupPort(final String id, final String name, final ProcessGroup processGroup, final RemoteProcessGroup remoteGroup,
            final TransferDirection direction, final ConnectableType type, final SSLContext sslContext, final ProcessScheduler scheduler,
        final NiFiProperties nifiProperties); @Override boolean isTargetRunning(); void setTargetRunning(final boolean targetRunning); @Override boolean isTriggerWhenEmpty(); @Override Resource getResource(); @Override Authorizable getParentAuthorizable(); @Override void shutdown(); @Override void onSchedulingStart(); @Override void onTrigger(final ProcessContext context, final ProcessSession session); @Override String getYieldPeriod(); @Override boolean getTargetExists(); @Override boolean isValid(); @Override Collection<ValidationResult> getValidationErrors(); @Override void verifyCanStart(); @Override void setUseCompression(final boolean useCompression); @Override boolean isUseCompression(); @Override String toString(); @Override RemoteProcessGroup getRemoteProcessGroup(); @Override TransferDirection getTransferDirection(); void setTargetExists(final boolean exists); @Override void removeConnection(final Connection connection); @Override SchedulingStrategy getSchedulingStrategy(); @Override boolean isSideEffectFree(); @Override String getComponentType(); static final String USER_AGENT; static final String CONTENT_TYPE; static final int GZIP_COMPRESSION_LEVEL; }