Apache
/
nifi
mirror of https://github.com/apache/nifi.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

NIFI-4391 Ensuring channel is closed when unable to connect in SocketChannelSender 

NIFI-4391 Adding debug logging of client port upon connection

Signed-off-by: Pierre Villard <pierre.villard.fr@gmail.com>

This closes #2159.
This commit is contained in:
Bryan Bende 2017-09-18 10:01:16 -04:00 committed by Pierre Villard
parent 7b07eb0577
commit a813ae113e
2 changed files with 45 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
nifi-nar-bundles/nifi-extension-utils/nifi-processor-utils/src/main/java/org/apache/nifi/processor/util/put/sender/SocketChannelSender.java
View File

@ -24,6 +24,7 @@ import java.io.IOException;
import java.io.OutputStream;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.net.SocketTimeoutException;
import java.net.StandardSocketOptions;
import java.nio.channels.SocketChannel;
@ -42,40 +43,53 @@ public class SocketChannelSender extends ChannelSender {
@Override
public void open() throws IOException {
if (channel == null) {
channel = SocketChannel.open();
channel.configureBlocking(false);
try {
if (channel == null) {
channel = SocketChannel.open();
channel.configureBlocking(false);
if (maxSendBufferSize > 0) {
channel.setOption(StandardSocketOptions.SO_SNDBUF, maxSendBufferSize);
final int actualSendBufSize = channel.getOption(StandardSocketOptions.SO_SNDBUF);
if (actualSendBufSize < maxSendBufferSize) {
logger.warn("Attempted to set Socket Send Buffer Size to " + maxSendBufferSize
+ " bytes but could only set to " + actualSendBufSize + "bytes. You may want to "
+ "consider changing the Operating System's maximum send buffer");
}
}
}
if (!channel.isConnected()) {
final long startTime = System.currentTimeMillis();
final InetSocketAddress socketAddress = new InetSocketAddress(InetAddress.getByName(host), port);
if (!channel.connect(socketAddress)) {
while (!channel.finishConnect()) {
if (System.currentTimeMillis() > startTime + timeout) {
throw new SocketTimeoutException("Timed out connecting to " + host + ":" + port);
}
try {
Thread.sleep(50L);
} catch (final InterruptedException e) {
if (maxSendBufferSize > 0) {
channel.setOption(StandardSocketOptions.SO_SNDBUF, maxSendBufferSize);
final int actualSendBufSize = channel.getOption(StandardSocketOptions.SO_SNDBUF);
if (actualSendBufSize < maxSendBufferSize) {
logger.warn("Attempted to set Socket Send Buffer Size to " + maxSendBufferSize
+ " bytes but could only set to " + actualSendBufSize + "bytes. You may want to "
+ "consider changing the Operating System's maximum send buffer");
}
}
}
socketChannelOutput = new SocketChannelOutputStream(channel);
socketChannelOutput.setTimeout(timeout);
if (!channel.isConnected()) {
final long startTime = System.currentTimeMillis();
final InetSocketAddress socketAddress = new InetSocketAddress(InetAddress.getByName(host), port);
if (!channel.connect(socketAddress)) {
while (!channel.finishConnect()) {
if (System.currentTimeMillis() > startTime + timeout) {
throw new SocketTimeoutException("Timed out connecting to " + host + ":" + port);
}
try {
Thread.sleep(50L);
} catch (final InterruptedException e) {
}
}
}
if (logger.isDebugEnabled()) {
final SocketAddress localAddress = channel.getLocalAddress();
if (localAddress != null && localAddress instanceof InetSocketAddress) {
final InetSocketAddress inetSocketAddress = (InetSocketAddress) localAddress;
logger.debug("Connected to local port {}", new Object[] {inetSocketAddress.getPort()});
}
}
socketChannelOutput = new SocketChannelOutputStream(channel);
socketChannelOutput.setTimeout(timeout);
}
} catch (final IOException e) {
IOUtils.closeQuietly(channel);
throw e;
}
}

3
nifi-nar-bundles/nifi-framework-bundle/nifi-framework/nifi-web/nifi-web-api/src/main/java/org/apache/nifi/web/api/ProcessGroupResource.java
View File

@ -1305,7 +1305,8 @@ public class ProcessGroupResource extends ApplicationResource {
}
private void activateControllerServices(final String groupId, final URI originalUri, final VariableRegistryUpdateRequest updateRequest,
final Pause pause, final Collection<AffectedComponentDTO> affectedServices, final ControllerServiceState desiredState, final VariableRegistryUpdateStep updateStep) throws InterruptedException {
final Pause pause, final Collection<AffectedComponentDTO> affectedServices, final ControllerServiceState desiredState, final VariableRegistryUpdateStep updateStep)
throws InterruptedException {
final Set<String> affectedServiceIds = affectedServices.stream()
.map(component -> component.getId())