You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by pnowojski <gi...@git.apache.org> on 2017/08/03 12:35:08 UTC
[GitHub] flink pull request #4470: [FLINK-7343] Simulate network failures in kafka at...
GitHub user pnowojski opened a pull request:
https://github.com/apache/flink/pull/4470
[FLINK-7343] Simulate network failures in kafka at-least-once test
We shouldn't fail KafkaServers directly, because they might not be able to flush the data (`log.flush.interval.***` properties). Since we don't want to test how well Kafka implements at-least-once/exactly-once semantic, it is a better idea (and hopefully more reliable) to just simulate network failure between Flink and Kafka in our at-least-once tests. To achieve that I have implemented `NetworkFailuresProxy` class.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/pnowojski/flink network-failures
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/4470.patch
To close this pull request, make a commit to your master/trunk branch
with (at least) the following in the commit message:
This closes #4470
----
commit 0e28327619893cfbf793fa842be3d965f649516c
Author: Piotr Nowojski <pi...@gmail.com>
Date: 2017-08-01T14:05:49Z
[FLINK-7343][kafka] Increase Xmx for tests
Sometimes 1000m was not enough memory to run at-least-once tests with broker failures on Travis
commit 8d820c3d0e77624a945e074f4a1bc476b5fd0f75
Author: Piotr Nowojski <pi...@gmail.com>
Date: 2017-08-01T16:11:27Z
[FLINK-7343] Add network proxy utility to simulate network failures
commit 967e1dfc87846b4011652bbaefab696900abc8dd
Author: Piotr Nowojski <pi...@gmail.com>
Date: 2017-08-03T07:25:04Z
fixup! [FLINK-7343][kafka] Increase Xmx for tests
commit 27b20f2ec3770231d95c3c7918c9313ce58b5e18
Author: Piotr Nowojski <pi...@gmail.com>
Date: 2017-08-03T09:27:12Z
[FLINK-7343] Use NetworkFailureProxy in kafka tests
We shouldn't fail KafkaServers directly, because they might not be able
to flush the data. Since we don't want to test how well Kafka implements
at-least-once/exactly-once semantic, we just simulate network failure
between Flink and Kafka in our at-least-once tests.
commit 692b5944f16b98aafe716ca1d18a04fa8a033798
Author: Piotr Nowojski <pi...@gmail.com>
Date: 2017-08-03T09:35:26Z
[hotfix][Kafka] Clean up getKafkaServer method
----
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4470: [FLINK-7343] Simulate network failures in kafka at...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4470#discussion_r131608115
--- Diff: flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java ---
@@ -0,0 +1,176 @@
+/*
+ * 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.flink.networking;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
+import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * Handler that is forwarding inbound traffic from the source channel to the target channel on remoteHost:remotePort
+ * and the responses in the opposite direction. All of the network traffic can be blocked at any time using blocked
+ * flag.
+ */
+class NetworkFailureHandler extends SimpleChannelUpstreamHandler {
+ private static final String TARGET_CHANNEL_HANDLER_NAME = "target_channel_handler";
+
+ // mapping between source and target channels, used for finding correct target channel to use for given source.
+ private final Map<Channel, Channel> sourceToTargetChannels = new ConcurrentHashMap<>();
+ private final Consumer<NetworkFailureHandler> onClose;
+ private final ClientSocketChannelFactory channelFactory;
+ private final String remoteHost;
+ private final int remotePort;
+
+ private final AtomicBoolean blocked;
+
+ public NetworkFailureHandler(
+ AtomicBoolean blocked,
+ Consumer<NetworkFailureHandler> onClose,
+ ClientSocketChannelFactory channelFactory,
+ String remoteHost,
+ int remotePort) {
+ this.blocked = blocked;
+ this.onClose = onClose;
+ this.channelFactory = channelFactory;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ }
+
+ /**
+ * Closes the specified channel after all queued write requests are flushed.
+ */
+ static void closeOnFlush(Channel channel) {
+ if (channel.isConnected()) {
+ channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ public void closeConnections() {
+ for (Map.Entry<Channel, Channel> entry : sourceToTargetChannels.entrySet()) {
+ // target channel is closed on source's channel channelClosed even
+ entry.getKey().close();
+ }
+ }
+
+ @Override
+ public void channelOpen(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ // Suspend incoming traffic until connected to the remote host.
+ final Channel sourceChannel = event.getChannel();
+ sourceChannel.setReadable(false);
+
+ if (blocked.get()) {
+ sourceChannel.close();
+ return;
+ }
+
+ // Start the connection attempt.
+ ClientBootstrap targetConnectionBootstrap = new ClientBootstrap(channelFactory);
+ targetConnectionBootstrap.getPipeline().addLast(TARGET_CHANNEL_HANDLER_NAME, new TargetChannelHandler(event.getChannel(), blocked));
+ ChannelFuture connectFuture = targetConnectionBootstrap.connect(new InetSocketAddress(remoteHost, remotePort));
+ sourceToTargetChannels.put(sourceChannel, connectFuture.getChannel());
+
+ connectFuture.addListener(future -> {
+ if (future.isSuccess()) {
+ // Connection attempt succeeded:
+ // Begin to accept incoming traffic.
+ sourceChannel.setReadable(true);
+ } else {
+ // Close the connection if the connection attempt has failed.
+ sourceChannel.close();
+ }
+ });
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext context, MessageEvent event) throws Exception {
+ if (blocked.get()) {
+ return;
+ }
+
+ ChannelBuffer msg = (ChannelBuffer) event.getMessage();
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ throw new IllegalStateException();
+ }
+ targetChannel.write(msg);
+ }
+
+ @Override
+ public void channelClosed(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ return;
+ }
+ closeOnFlush(targetChannel);
+ sourceToTargetChannels.remove(event.getChannel());
+ onClose.accept(this);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) throws Exception {
+ event.getCause().printStackTrace();
--- End diff --
I suggest to replace the `printStackTrace()` with logging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4470: [FLINK-7343] Simulate network failures in kafka at-least-...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4470
Please close the commit and the Jira issue.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4470: [FLINK-7343] Simulate network failures in kafka at...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on a diff in the pull request:
https://github.com/apache/flink/pull/4470#discussion_r131648638
--- Diff: flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java ---
@@ -0,0 +1,176 @@
+/*
+ * 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.flink.networking;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
+import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * Handler that is forwarding inbound traffic from the source channel to the target channel on remoteHost:remotePort
+ * and the responses in the opposite direction. All of the network traffic can be blocked at any time using blocked
+ * flag.
+ */
+class NetworkFailureHandler extends SimpleChannelUpstreamHandler {
+ private static final String TARGET_CHANNEL_HANDLER_NAME = "target_channel_handler";
+
+ // mapping between source and target channels, used for finding correct target channel to use for given source.
+ private final Map<Channel, Channel> sourceToTargetChannels = new ConcurrentHashMap<>();
+ private final Consumer<NetworkFailureHandler> onClose;
+ private final ClientSocketChannelFactory channelFactory;
+ private final String remoteHost;
+ private final int remotePort;
+
+ private final AtomicBoolean blocked;
+
+ public NetworkFailureHandler(
+ AtomicBoolean blocked,
+ Consumer<NetworkFailureHandler> onClose,
+ ClientSocketChannelFactory channelFactory,
+ String remoteHost,
+ int remotePort) {
+ this.blocked = blocked;
+ this.onClose = onClose;
+ this.channelFactory = channelFactory;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ }
+
+ /**
+ * Closes the specified channel after all queued write requests are flushed.
+ */
+ static void closeOnFlush(Channel channel) {
+ if (channel.isConnected()) {
+ channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ public void closeConnections() {
+ for (Map.Entry<Channel, Channel> entry : sourceToTargetChannels.entrySet()) {
+ // target channel is closed on source's channel channelClosed even
+ entry.getKey().close();
+ }
+ }
+
+ @Override
+ public void channelOpen(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ // Suspend incoming traffic until connected to the remote host.
+ final Channel sourceChannel = event.getChannel();
+ sourceChannel.setReadable(false);
+
+ if (blocked.get()) {
+ sourceChannel.close();
+ return;
+ }
+
+ // Start the connection attempt.
+ ClientBootstrap targetConnectionBootstrap = new ClientBootstrap(channelFactory);
+ targetConnectionBootstrap.getPipeline().addLast(TARGET_CHANNEL_HANDLER_NAME, new TargetChannelHandler(event.getChannel(), blocked));
+ ChannelFuture connectFuture = targetConnectionBootstrap.connect(new InetSocketAddress(remoteHost, remotePort));
+ sourceToTargetChannels.put(sourceChannel, connectFuture.getChannel());
+
+ connectFuture.addListener(future -> {
+ if (future.isSuccess()) {
+ // Connection attempt succeeded:
+ // Begin to accept incoming traffic.
+ sourceChannel.setReadable(true);
+ } else {
+ // Close the connection if the connection attempt has failed.
+ sourceChannel.close();
+ }
+ });
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext context, MessageEvent event) throws Exception {
+ if (blocked.get()) {
+ return;
+ }
+
+ ChannelBuffer msg = (ChannelBuffer) event.getMessage();
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ throw new IllegalStateException();
+ }
+ targetChannel.write(msg);
+ }
+
+ @Override
+ public void channelClosed(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ return;
+ }
+ closeOnFlush(targetChannel);
+ sourceToTargetChannels.remove(event.getChannel());
+ onClose.accept(this);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) throws Exception {
+ event.getCause().printStackTrace();
--- End diff --
Ops, missed that when cleaning up a copied example code.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4470: [FLINK-7343] Simulate network failures in kafka at...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski closed the pull request at:
https://github.com/apache/flink/pull/4470
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4470: [FLINK-7343] Simulate network failures in kafka at-least-...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on the issue:
https://github.com/apache/flink/pull/4470
Thanks :)
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4470: [FLINK-7343] Simulate network failures in kafka at-least-...
Posted by pnowojski <gi...@git.apache.org>.
Github user pnowojski commented on the issue:
https://github.com/apache/flink/pull/4470
Yes I also didn't like adding this new flag, but didn't have enough motivation to change it. I have done some refactoring extracting those dynamically set in `prepare` method to some `Config` class.
However it helps only a little bit. Those tests would need a more comprehensive refactor in the future. I particularly don't like that this `prepare` method exists, it should all be configured in the constructor and all of those should be final fields.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4470: [FLINK-7343] Simulate network failures in kafka at...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4470#discussion_r131608304
--- Diff: flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java ---
@@ -0,0 +1,176 @@
+/*
+ * 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.flink.networking;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
+import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * Handler that is forwarding inbound traffic from the source channel to the target channel on remoteHost:remotePort
+ * and the responses in the opposite direction. All of the network traffic can be blocked at any time using blocked
+ * flag.
+ */
+class NetworkFailureHandler extends SimpleChannelUpstreamHandler {
+ private static final String TARGET_CHANNEL_HANDLER_NAME = "target_channel_handler";
+
+ // mapping between source and target channels, used for finding correct target channel to use for given source.
+ private final Map<Channel, Channel> sourceToTargetChannels = new ConcurrentHashMap<>();
+ private final Consumer<NetworkFailureHandler> onClose;
+ private final ClientSocketChannelFactory channelFactory;
+ private final String remoteHost;
+ private final int remotePort;
+
+ private final AtomicBoolean blocked;
+
+ public NetworkFailureHandler(
+ AtomicBoolean blocked,
+ Consumer<NetworkFailureHandler> onClose,
+ ClientSocketChannelFactory channelFactory,
+ String remoteHost,
+ int remotePort) {
+ this.blocked = blocked;
+ this.onClose = onClose;
+ this.channelFactory = channelFactory;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ }
+
+ /**
+ * Closes the specified channel after all queued write requests are flushed.
+ */
+ static void closeOnFlush(Channel channel) {
+ if (channel.isConnected()) {
+ channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ public void closeConnections() {
+ for (Map.Entry<Channel, Channel> entry : sourceToTargetChannels.entrySet()) {
+ // target channel is closed on source's channel channelClosed even
+ entry.getKey().close();
+ }
+ }
+
+ @Override
+ public void channelOpen(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ // Suspend incoming traffic until connected to the remote host.
+ final Channel sourceChannel = event.getChannel();
+ sourceChannel.setReadable(false);
+
+ if (blocked.get()) {
+ sourceChannel.close();
+ return;
+ }
+
+ // Start the connection attempt.
+ ClientBootstrap targetConnectionBootstrap = new ClientBootstrap(channelFactory);
+ targetConnectionBootstrap.getPipeline().addLast(TARGET_CHANNEL_HANDLER_NAME, new TargetChannelHandler(event.getChannel(), blocked));
+ ChannelFuture connectFuture = targetConnectionBootstrap.connect(new InetSocketAddress(remoteHost, remotePort));
+ sourceToTargetChannels.put(sourceChannel, connectFuture.getChannel());
+
+ connectFuture.addListener(future -> {
+ if (future.isSuccess()) {
+ // Connection attempt succeeded:
+ // Begin to accept incoming traffic.
+ sourceChannel.setReadable(true);
+ } else {
+ // Close the connection if the connection attempt has failed.
+ sourceChannel.close();
+ }
+ });
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext context, MessageEvent event) throws Exception {
+ if (blocked.get()) {
+ return;
+ }
+
+ ChannelBuffer msg = (ChannelBuffer) event.getMessage();
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ throw new IllegalStateException();
--- End diff --
A cause message could be added.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink issue #4470: [FLINK-7343] Simulate network failures in kafka at-least-...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on the issue:
https://github.com/apache/flink/pull/4470
Merging.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---
[GitHub] flink pull request #4470: [FLINK-7343] Simulate network failures in kafka at...
Posted by StefanRRichter <gi...@git.apache.org>.
Github user StefanRRichter commented on a diff in the pull request:
https://github.com/apache/flink/pull/4470#discussion_r131609166
--- Diff: flink-test-utils-parent/flink-test-utils/src/main/java/org/apache/flink/networking/NetworkFailureHandler.java ---
@@ -0,0 +1,176 @@
+/*
+ * 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.flink.networking;
+
+import org.jboss.netty.bootstrap.ClientBootstrap;
+import org.jboss.netty.buffer.ChannelBuffer;
+import org.jboss.netty.buffer.ChannelBuffers;
+import org.jboss.netty.channel.Channel;
+import org.jboss.netty.channel.ChannelFuture;
+import org.jboss.netty.channel.ChannelFutureListener;
+import org.jboss.netty.channel.ChannelHandlerContext;
+import org.jboss.netty.channel.ChannelStateEvent;
+import org.jboss.netty.channel.ExceptionEvent;
+import org.jboss.netty.channel.MessageEvent;
+import org.jboss.netty.channel.SimpleChannelUpstreamHandler;
+import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
+
+import java.net.InetSocketAddress;
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+import java.util.concurrent.atomic.AtomicBoolean;
+import java.util.function.Consumer;
+
+/**
+ * Handler that is forwarding inbound traffic from the source channel to the target channel on remoteHost:remotePort
+ * and the responses in the opposite direction. All of the network traffic can be blocked at any time using blocked
+ * flag.
+ */
+class NetworkFailureHandler extends SimpleChannelUpstreamHandler {
+ private static final String TARGET_CHANNEL_HANDLER_NAME = "target_channel_handler";
+
+ // mapping between source and target channels, used for finding correct target channel to use for given source.
+ private final Map<Channel, Channel> sourceToTargetChannels = new ConcurrentHashMap<>();
+ private final Consumer<NetworkFailureHandler> onClose;
+ private final ClientSocketChannelFactory channelFactory;
+ private final String remoteHost;
+ private final int remotePort;
+
+ private final AtomicBoolean blocked;
+
+ public NetworkFailureHandler(
+ AtomicBoolean blocked,
+ Consumer<NetworkFailureHandler> onClose,
+ ClientSocketChannelFactory channelFactory,
+ String remoteHost,
+ int remotePort) {
+ this.blocked = blocked;
+ this.onClose = onClose;
+ this.channelFactory = channelFactory;
+ this.remoteHost = remoteHost;
+ this.remotePort = remotePort;
+ }
+
+ /**
+ * Closes the specified channel after all queued write requests are flushed.
+ */
+ static void closeOnFlush(Channel channel) {
+ if (channel.isConnected()) {
+ channel.write(ChannelBuffers.EMPTY_BUFFER).addListener(ChannelFutureListener.CLOSE);
+ }
+ }
+
+ public void closeConnections() {
+ for (Map.Entry<Channel, Channel> entry : sourceToTargetChannels.entrySet()) {
+ // target channel is closed on source's channel channelClosed even
+ entry.getKey().close();
+ }
+ }
+
+ @Override
+ public void channelOpen(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ // Suspend incoming traffic until connected to the remote host.
+ final Channel sourceChannel = event.getChannel();
+ sourceChannel.setReadable(false);
+
+ if (blocked.get()) {
+ sourceChannel.close();
+ return;
+ }
+
+ // Start the connection attempt.
+ ClientBootstrap targetConnectionBootstrap = new ClientBootstrap(channelFactory);
+ targetConnectionBootstrap.getPipeline().addLast(TARGET_CHANNEL_HANDLER_NAME, new TargetChannelHandler(event.getChannel(), blocked));
+ ChannelFuture connectFuture = targetConnectionBootstrap.connect(new InetSocketAddress(remoteHost, remotePort));
+ sourceToTargetChannels.put(sourceChannel, connectFuture.getChannel());
+
+ connectFuture.addListener(future -> {
+ if (future.isSuccess()) {
+ // Connection attempt succeeded:
+ // Begin to accept incoming traffic.
+ sourceChannel.setReadable(true);
+ } else {
+ // Close the connection if the connection attempt has failed.
+ sourceChannel.close();
+ }
+ });
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext context, MessageEvent event) throws Exception {
+ if (blocked.get()) {
+ return;
+ }
+
+ ChannelBuffer msg = (ChannelBuffer) event.getMessage();
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ throw new IllegalStateException();
+ }
+ targetChannel.write(msg);
+ }
+
+ @Override
+ public void channelClosed(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ Channel targetChannel = sourceToTargetChannels.get(event.getChannel());
+ if (targetChannel == null) {
+ return;
+ }
+ closeOnFlush(targetChannel);
+ sourceToTargetChannels.remove(event.getChannel());
+ onClose.accept(this);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event) throws Exception {
+ event.getCause().printStackTrace();
+ closeOnFlush(event.getChannel());
+ }
+
+ private static class TargetChannelHandler extends SimpleChannelUpstreamHandler {
+ private final Channel sourceChannel;
+ private final AtomicBoolean blocked;
+
+ TargetChannelHandler(Channel sourceChannel, AtomicBoolean blocked) {
+ this.sourceChannel = sourceChannel;
+ this.blocked = blocked;
+ }
+
+ @Override
+ public void messageReceived(ChannelHandlerContext context, MessageEvent event) throws Exception {
+ if (blocked.get()) {
+ return;
+ }
+ ChannelBuffer msg = (ChannelBuffer) event.getMessage();
+ sourceChannel.write(msg);
+ }
+
+ @Override
+ public void channelClosed(ChannelHandlerContext context, ChannelStateEvent event) throws Exception {
+ closeOnFlush(sourceChannel);
+ }
+
+ @Override
+ public void exceptionCaught(ChannelHandlerContext context, ExceptionEvent event)
+ throws Exception {
+ event.getCause().printStackTrace();
--- End diff --
Same suggestion to use logging over `printStackTrace ()`.
---
If your project is set up for it, you can reply to this email and have your
reply appear on GitHub as well. If your project does not have this feature
enabled and wishes so, or if the feature is enabled but not working, please
contact infrastructure at infrastructure@apache.org or file a JIRA ticket
with INFRA.
---