[ https://issues.apache.org/jira/browse/FLINK-2536?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=14709175#comment-14709175 ]
ASF GitHub Bot commented on FLINK-2536: --------------------------------------- Github user tillrohrmann commented on a diff in the pull request: https://github.com/apache/flink/pull/1030#discussion_r37745586 --- Diff: flink-staging/flink-streaming/flink-streaming-core/src/test/java/org/apache/flink/streaming/api/functions/sink/SocketClientSinkTest.java --- @@ -0,0 +1,301 @@ +/* + * 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.streaming.api.functions.sink; + +import java.io.BufferedReader; +import java.io.InputStreamReader; +import java.net.Socket; + +import org.apache.flink.configuration.Configuration; +import org.apache.flink.streaming.util.serialization.SerializationSchema; + +import org.junit.Test; + +import static org.junit.Assert.assertEquals; +import static org.junit.Assert.assertTrue; +import static org.junit.Assert.fail; + +import java.net.ServerSocket; +import java.util.concurrent.atomic.AtomicReference; + +/** + * Tests for the {@link org.apache.flink.streaming.api.functions.sink.SocketClientSink}. + */ +public class SocketClientSinkTest{ + + final AtomicReference<Throwable> error = new AtomicReference<Throwable>(); + private final String host = "127.0.0.1"; + private int port; + private String value; + + public Thread t; + + public SocketClientSinkTest() { + } + + @Test + public void testSocketSink() throws Exception{ + value = ""; + ServerSocket server = new ServerSocket(0); + port = server.getLocalPort(); + + new Thread(new Runnable() { + @Override + public void run() { + t = Thread.currentThread(); + SerializationSchema<String, byte[]> simpleSchema = new SerializationSchema<String, byte[]>() { + @Override + public byte[] serialize(String element) { + return element.getBytes(); + } + }; + + try { + SocketClientSink<String> simpleSink = new SocketClientSink<String>(host, port, simpleSchema, 0); + simpleSink.open(new Configuration()); + simpleSink.invoke("testSocketSinkInvoke"); + simpleSink.close(); + } catch (Exception e){ + error.set(e); + } + } + }).start(); + + Socket sk = server.accept(); + BufferedReader rdr = new BufferedReader(new InputStreamReader(sk + .getInputStream())); + value = rdr.readLine(); + + t.join(); + server.close(); + + if (error.get() != null) { + Throwable t = error.get(); + t.printStackTrace(); + fail("Error in spawned thread: " + t.getMessage()); + } + + assertEquals("testSocketSinkInvoke", value); + } + + @Test + public void testSocketSinkNoRetry() throws Exception{ + ServerSocket server = new ServerSocket(0); + port = server.getLocalPort(); + + new Thread(new Runnable() { + + @Override + public void run() { + t = Thread.currentThread(); + SerializationSchema<String, byte[]> simpleSchema = new SerializationSchema<String, byte[]>() { + @Override + public byte[] serialize(String element) { + return element.getBytes(); + } + }; + + try { + SocketClientSink<String> simpleSink = new SocketClientSink<String>(host, port, simpleSchema, 0); + simpleSink.open(new Configuration()); + + synchronized (t) { + //wating for server to close + t.wait(); + } + + //firstly send a fin to cancel the client state:FIN_WAIT_2 while the server is CLOSE_WAIT + simpleSink.invoke("testSocketSinkInvoke"); + + //socket is closed then test "retry" + simpleSink.invoke("testSocketSinkInvoke"); + simpleSink.close(); + assertEquals(0, simpleSink.retries); + } catch (Exception e) { + //This Exception is good since the socket server is never reopen. + } + } + }).start(); + + Socket sk = server.accept(); + sk.setKeepAlive(false); + sk.close(); + server.close(); + + synchronized (t) { + t.notifyAll(); + } + + t.join(); + if (error.get() != null) { + Throwable t = error.get(); + t.printStackTrace(); + fail("Error in spawned thread: " + t.getMessage()); + } + } + + @Test + public void testSocketSinkRetryTenTimes() throws Exception{ + ServerSocket server = new ServerSocket(0); + port = server.getLocalPort(); + + new Thread(new Runnable() { + + @Override + public void run() { + t = Thread.currentThread(); + SerializationSchema<String, byte[]> simpleSchema = new SerializationSchema<String, byte[]>() { + @Override + public byte[] serialize(String element) { + return element.getBytes(); + } + }; + + try { + SocketClientSink<String> simpleSink = new SocketClientSink<String>(host, port, simpleSchema, 10); + simpleSink.open(new Configuration()); + + synchronized (t){ + //wating for server to close + t.wait(); + } + + //firstly send a fin to cancel the client state:FIN_WAIT_2 while the server is CLOSE_WAIT + simpleSink.invoke("testSocketSinkInvoke"); + + //socket is closed then test "retry" + simpleSink.invoke("testSocketSinkInvoke"); + simpleSink.close(); + assertEquals(10, simpleSink.retries); + } catch (Exception e) { + //This Exception is good since the socket server is never reopen. + } + } + }).start(); + + Socket sk = server.accept(); + sk.setKeepAlive(false); + sk.close(); + server.close(); + + synchronized (t) { + t.notifyAll(); + } + + t.join(); + if (error.get() != null) { + Throwable t = error.get(); + t.printStackTrace(); + fail("Error in spawned thread: " + t.getMessage()); + } + } + + @Test + public void testSocketSinkRetryAccess() throws Exception{ + ServerSocket server = new ServerSocket(0); + port = server.getLocalPort(); + + new Thread(new Runnable() { + + Thread t1; + SocketClientSink<String> simpleSink; + + @Override + public void run() { + t = Thread.currentThread(); + SerializationSchema<String, byte[]> simpleSchema = new SerializationSchema<String, byte[]>() { + @Override + public byte[] serialize(String element) { + return element.getBytes(); + } + }; + + try { + simpleSink = new SocketClientSink<String>(host, port, simpleSchema, 10); + simpleSink.open(new Configuration()); + + synchronized (t) { + //wating for server to close + t.wait(); + } + + //firstly send a fin to cancel the client state:FIN_WAIT_2 while the server is CLOSE_WAIT + simpleSink.invoke("testSocketSinkInvoke"); + + new Thread(new Runnable() { + + @Override + public void run() { + t1 = Thread.currentThread(); + try { + //socket is closed then test "retry" + simpleSink.invoke("testSocketSinkInvoke"); + simpleSink.close(); + } catch (Exception e) { + error.set(e); + } + } + + }).start(); + + //set a new server to let the retry success. + while (simpleSink.retries == 0){ + synchronized (simpleSink) { + simpleSink.notifyAll(); + } + } + + //reopen socket server for sink access + value = ""; + ServerSocket server = new ServerSocket(port); --- End diff -- How do you make sure that the `SocketClientSink` has not been terminated before you open the server socket? > Add a retry for SocketClientSink > -------------------------------- > > Key: FLINK-2536 > URL: https://issues.apache.org/jira/browse/FLINK-2536 > Project: Flink > Issue Type: Improvement > Components: Streaming > Affects Versions: 0.10 > Reporter: Huang Wei > Fix For: 0.10 > > Original Estimate: 168h > Remaining Estimate: 168h > > I found the SocketClientSink doesn`t use a re-connect when disconnect from > the socket server or get exception. > I`d like to add a re-connect like socket source for socket sink. -- This message was sent by Atlassian JIRA (v6.3.4#6332)