You are viewing a plain text version of this content. The canonical link for it is here.
Posted to issues@flink.apache.org by kl0u <gi...@git.apache.org> on 2016/07/14 09:20:47 UTC
[GitHub] flink pull request #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
GitHub user kl0u opened a pull request:
https://github.com/apache/flink/pull/2245
[hotfix] Fix Redis Sink to fail at opening if Redis is not initialized.
Before if Redis was not initialized, the job would start and would wait until the first element arrives in order to realize that Redis is not running. This fixes that by checking the connection to redis during the opening phase.
You can merge this pull request into a Git repository by running:
$ git pull https://github.com/kl0u/flink redis_check
Alternatively you can review and apply these changes as the patch at:
https://github.com/apache/flink/pull/2245.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 #2245
----
commit 9a31902de93732d2e3c39d3e23ed3e34162795ec
Author: kl0u <kk...@gmail.com>
Date: 2016-07-13T16:40:17Z
[hotfix] Fix Redis Sink to fail at opening if Redis is not initialized.
----
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71869449
--- Diff: flink-streaming-connectors/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisSinkTest.java ---
@@ -105,18 +105,13 @@ private void testDownBehavior(FlinkJedisConfigBase config) throws Exception {
try {
redisSink.open(new Configuration());
} catch (Throwable e) {
- Throwable init = e;
- // search for nested JedisConnectionExceptions
+ // search for the JedisConnectionExceptions
// because this is the expected behavior
-
- int depth = 0;
- while (!(e instanceof JedisConnectionException)) {
- e = e.getCause();
- if (e == null || depth++ == 20) {
- init.printStackTrace();
- fail("Test Failed: " + init.getMessage());
- }
+
+ if (!(e instanceof JedisConnectionException)) {
+ e.printStackTrace();
+ fail("Test Failed: " + e.getMessage());
--- End diff --
I think it is better to let the exception bubble up instead of printing the stack trace and calling `fail`.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redis is no...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/2245
I think this is ok now.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71698908
--- Diff: flink-streaming-connectors/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisSinkTest.java ---
@@ -103,15 +105,17 @@ private void testDownBehavior(FlinkJedisConfigBase config) throws Exception {
try {
redisSink.open(new Configuration());
} catch (Throwable e) {
+ Throwable init = e;
- // search for nested ConnectionExceptions
+ // search for nested JedisConnectionExceptions
// because this is the expected behavior
int depth = 0;
while (!(e instanceof JedisConnectionException)) {
- Throwable cause = e.getCause();
- if (cause == null || depth++ == 20) {
- throw e;
+ e = e.getCause();
+ if (e == null || depth++ == 20) {
+ init.printStackTrace();
+ fail("Test Failed: " + init.getMessage());
--- End diff --
Why don't we simply throw the initial exception here?
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71156944
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/RedisSink.java ---
@@ -166,7 +166,13 @@ public void invoke(IN input) throws Exception {
*/
@Override
public void open(Configuration parameters) throws Exception {
- this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
+ try {
+ this.redisCommandsContainer = RedisCommandsContainerBuilder.build(this.flinkJedisConfigBase);
+ this.redisCommandsContainer.open();
+ } catch (Exception e) {
+ LOG.error("Redis has not been properly initialized.");
--- End diff --
The exception should be logged here as well.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71157753
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java ---
@@ -68,12 +68,12 @@ public RedisContainer(final JedisSentinelPool sentinelPool) {
*/
@Override
public void close() throws IOException {
- if (this.jedisPool != null) {
- this.jedisPool.close();
- }
- if (this.jedisSentinelPool != null) {
- this.jedisSentinelPool.close();
- }
+ getInstance().close();
+ }
+
+ @Override
+ public void open() throws Exception {
+ getInstance().echo("Test");
--- End diff --
Yes. It tries to open a connection and just echo the message "Test".
This is to see if there is a way to communicate with the cluster or server (depending on the setup).
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redis is no...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/2245
Thanks a lot @tillrohrmann !
I integrated your comment.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redis is no...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2245
Thanks for your contribution @kl0u. I had some inline comments.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71379345
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java ---
@@ -68,12 +68,12 @@ public RedisContainer(final JedisSentinelPool sentinelPool) {
*/
@Override
public void close() throws IOException {
- if (this.jedisPool != null) {
- this.jedisPool.close();
- }
- if (this.jedisSentinelPool != null) {
- this.jedisSentinelPool.close();
- }
+ getInstance().close();
--- End diff --
Yes because from the constructors, only one of the `jedisSentinelPool` or `jedisPool` can be non-null value and that is what `getInstance()` returns.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71512375
--- Diff: flink-streaming-connectors/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisSinkTest.java ---
@@ -41,6 +50,73 @@ public void shouldThrowNullPointerExceptionIfConfigurationIsNull(){
new RedisSink<>(null, new TestMapper(new RedisCommandDescription(RedisCommand.LPUSH)));
}
+ @Test
+ public void testRedisDownBehavior() throws Exception {
+
+ // create a wrong configuration so that open() fails.
+
+ FlinkJedisPoolConfig wrongJedisPoolConfig = new FlinkJedisPoolConfig.Builder()
+ .setHost("127.0.0.1")
+ .setPort(1234).build();
+
+ testDownBehavior(wrongJedisPoolConfig);
+ }
+
+ @Test
+ public void testRedisClusterDownBehavior() throws Exception {
+
+ Set<InetSocketAddress> hosts = new HashSet<>();
+ hosts.add(new InetSocketAddress("127.0.0.1", 1234));
+
+ // create a wrong configuration so that open() fails.
+
+ FlinkJedisClusterConfig wrongJedisClusterConfig = new FlinkJedisClusterConfig.Builder()
+ .setNodes(hosts)
+ .setTimeout(100)
+ .setMaxIdle(1)
+ .setMaxTotal(1)
+ .setMinIdle(1).build();
+
+ testDownBehavior(wrongJedisClusterConfig);
+ }
+
+ @Test
+ public void testRedisSentinelDownBehavior() throws Exception {
+
+ Set<String> hosts = new HashSet<>();
+ hosts.add("localhost:55095");
+
+ // create a wrong configuration so that open() fails.
+
+ FlinkJedisSentinelConfig wrongJedisSentinelConfig = new FlinkJedisSentinelConfig.Builder()
+ .setMasterName("master")
+ .setSentinels(hosts)
+ .build();
+
+ testDownBehavior(wrongJedisSentinelConfig);
+ }
+
+ private void testDownBehavior(FlinkJedisConfigBase config) throws Exception {
+ RedisSink<Tuple2<String, String>> redisSink = new RedisSink<>(config,
+ new RedisSinkITCase.RedisCommandMapper(RedisCommand.SADD));
+
+ try {
+ redisSink.open(new Configuration());
+ } catch (Throwable e) {
+
+ // search for nested ConnectionExceptions
+ // because this is the expected behavior
+
+ int depth = 0;
+ while (!(e instanceof JedisConnectionException)) {
--- End diff --
This condition will always be true/false if you don't reset `e = cause` in the while loop body.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71157765
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java ---
@@ -68,12 +68,12 @@ public RedisContainer(final JedisSentinelPool sentinelPool) {
*/
@Override
public void close() throws IOException {
- if (this.jedisPool != null) {
- this.jedisPool.close();
- }
- if (this.jedisSentinelPool != null) {
- this.jedisSentinelPool.close();
- }
+ getInstance().close();
--- End diff --
Is this equivalent to the old `close` method definition?
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71869391
--- Diff: flink-streaming-connectors/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisSinkTest.java ---
@@ -105,18 +105,13 @@ private void testDownBehavior(FlinkJedisConfigBase config) throws Exception {
try {
redisSink.open(new Configuration());
} catch (Throwable e) {
- Throwable init = e;
- // search for nested JedisConnectionExceptions
+ // search for the JedisConnectionExceptions
// because this is the expected behavior
-
- int depth = 0;
- while (!(e instanceof JedisConnectionException)) {
- e = e.getCause();
- if (e == null || depth++ == 20) {
- init.printStackTrace();
- fail("Test Failed: " + init.getMessage());
- }
+
+ if (!(e instanceof JedisConnectionException)) {
+ e.printStackTrace();
+ fail("Test Failed: " + e.getMessage());
--- End diff --
Why did you change the `while` loop into a `if` condition. Did change something that the `JedisConnectionException` will always be the top level exception?
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71157239
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java ---
@@ -68,12 +68,12 @@ public RedisContainer(final JedisSentinelPool sentinelPool) {
*/
@Override
public void close() throws IOException {
- if (this.jedisPool != null) {
- this.jedisPool.close();
- }
- if (this.jedisSentinelPool != null) {
- this.jedisSentinelPool.close();
- }
+ getInstance().close();
+ }
+
+ @Override
+ public void open() throws Exception {
+ getInstance().echo("Test");
--- End diff --
`echo("Test")`?
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redis is no...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on the issue:
https://github.com/apache/flink/pull/2245
Changes look good @kl0u. Good work :-) There is only a minor issue with one of the test cases and its exception handling.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71159044
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java ---
@@ -68,12 +68,12 @@ public RedisContainer(final JedisSentinelPool sentinelPool) {
*/
@Override
public void close() throws IOException {
- if (this.jedisPool != null) {
- this.jedisPool.close();
- }
- if (this.jedisSentinelPool != null) {
- this.jedisSentinelPool.close();
- }
+ getInstance().close();
+ }
+
+ @Override
+ public void open() throws Exception {
+ getInstance().echo("Test");
--- End diff --
Is this the only way to test a connection? If so, then it should be documented properly that this is used to test the connection.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71158368
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisClusterContainer.java ---
@@ -47,6 +47,11 @@ public RedisClusterContainer(JedisCluster jedisCluster) {
}
--- End diff --
`jedisCluster` is not serializable
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71869640
--- Diff: flink-streaming-connectors/flink-connector-redis/src/test/java/org/apache/flink/streaming/connectors/redis/RedisSinkTest.java ---
@@ -105,18 +105,13 @@ private void testDownBehavior(FlinkJedisConfigBase config) throws Exception {
try {
redisSink.open(new Configuration());
} catch (Throwable e) {
- Throwable init = e;
- // search for nested JedisConnectionExceptions
+ // search for the JedisConnectionExceptions
// because this is the expected behavior
-
- int depth = 0;
- while (!(e instanceof JedisConnectionException)) {
- e = e.getCause();
- if (e == null || depth++ == 20) {
- init.printStackTrace();
- fail("Test Failed: " + init.getMessage());
- }
+
+ if (!(e instanceof JedisConnectionException)) {
+ e.printStackTrace();
+ fail("Test Failed: " + e.getMessage());
--- End diff --
I also thought so that is why I had it like this but I thought your previous comment was to remove it.
Probably i misunderstood. Actually the exception was in the first layer since the beginning.
I can change it back ;)
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redis is no...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/2245
Hi @tillrohrmann !Thanks for the comments, I addressed the ones that should be addressed.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redis is no...
Posted by kl0u <gi...@git.apache.org>.
Github user kl0u commented on the issue:
https://github.com/apache/flink/pull/2245
Done! Thanks @tillrohrmann
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by asfgit <gi...@git.apache.org>.
Github user asfgit closed the pull request at:
https://github.com/apache/flink/pull/2245
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71157963
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisContainer.java ---
@@ -68,12 +68,12 @@ public RedisContainer(final JedisSentinelPool sentinelPool) {
*/
--- End diff --
The `RedisContainer` class contains fields which are not serializable. The `jedisPool` and the `jedisSentinelPool`.
---
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 #2245: [hotfix] Fix Redis Sink to fail at opening if Redi...
Posted by tillrohrmann <gi...@git.apache.org>.
Github user tillrohrmann commented on a diff in the pull request:
https://github.com/apache/flink/pull/2245#discussion_r71157090
--- Diff: flink-streaming-connectors/flink-connector-redis/src/main/java/org/apache/flink/streaming/connectors/redis/common/container/RedisClusterContainer.java ---
@@ -47,6 +47,11 @@ public RedisClusterContainer(JedisCluster jedisCluster) {
}
@Override
+ public void open() throws Exception {
+ jedisCluster.echo("Test");
--- End diff --
testing artifact?
---
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.
---