You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@flink.apache.org by se...@apache.org on 2016/03/13 17:24:27 UTC
[1/4] flink git commit: [tests] Fix unstable test WebFrontendITCase
Repository: flink
Updated Branches:
refs/heads/master 0f8d76c6f -> 4870fd62f
[tests] Fix unstable test WebFrontendITCase
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/acb051d8
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/acb051d8
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/acb051d8
Branch: refs/heads/master
Commit: acb051d8db3acca0ff97628b602aa21dcbf3604f
Parents: 0f8d76c
Author: Stephan Ewen <se...@apache.org>
Authored: Fri Mar 11 18:57:13 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Mar 13 16:03:03 2016 +0100
----------------------------------------------------------------------
.../flink/test/web/WebFrontendITCase.java | 171 ++++++++++---------
1 file changed, 90 insertions(+), 81 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/acb051d8/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
----------------------------------------------------------------------
diff --git a/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java b/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
index 48888a4..31084a0 100644
--- a/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
+++ b/flink-tests/src/test/java/org/apache/flink/test/web/WebFrontendITCase.java
@@ -27,28 +27,26 @@ import org.apache.commons.io.FileUtils;
import org.apache.flink.api.common.ExecutionConfig;
import org.apache.flink.api.common.JobID;
import org.apache.flink.configuration.ConfigConstants;
+import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.jobgraph.JobGraph;
import org.apache.flink.runtime.jobgraph.JobVertex;
import org.apache.flink.runtime.testutils.StoppableInvokable;
-import org.apache.flink.runtime.webmonitor.WebMonitor;
import org.apache.flink.runtime.webmonitor.WebMonitorUtils;
import org.apache.flink.runtime.webmonitor.files.MimeTypes;
import org.apache.flink.runtime.webmonitor.testutils.HttpTestClient;
-import org.apache.flink.test.util.MultipleProgramsTestBase;
+import org.apache.flink.test.util.ForkableFlinkMiniCluster;
import org.apache.flink.test.util.TestBaseUtils;
-import org.apache.sling.commons.json.JSONArray;
-import org.apache.sling.commons.json.JSONObject;
+
import org.junit.BeforeClass;
import org.junit.Test;
-import org.junit.runner.RunWith;
-import org.junit.runners.Parameterized;
import scala.concurrent.duration.Deadline;
import scala.concurrent.duration.FiniteDuration;
+
import io.netty.handler.codec.http.HttpResponseStatus;
-import java.util.Arrays;
-import java.util.Collection;
+import java.io.File;
+import java.nio.file.Files;
import java.util.Map;
import java.util.concurrent.TimeUnit;
@@ -57,30 +55,40 @@ import static org.junit.Assert.assertNotNull;
import static org.junit.Assert.assertTrue;
import static org.junit.Assert.fail;
-@RunWith(Parameterized.class)
-public class WebFrontendITCase extends MultipleProgramsTestBase {
+import static org.apache.flink.test.util.TestBaseUtils.getFromHTTP;
- // make sure that the webserver is started for us!
- static {
- startWebServer = true;
- }
+public class WebFrontendITCase {
- private static int port = -1;
+ private static final int NUM_TASK_MANAGERS = 2;
+ private static final int NUM_SLOTS = 4;
+
+ private static ForkableFlinkMiniCluster cluster;
+ private static int port = -1;
+
@BeforeClass
- public static void initialize() {
- WebMonitor webMonitor = cluster.webMonitor().get();
- port = webMonitor.getServerPort();
- }
-
- public WebFrontendITCase(TestExecutionMode m) {
- super(m);
- }
-
- @Parameterized.Parameters(name = "Execution mode = {0}")
- public static Collection<Object[]> executionModes() {
- return Arrays.<Object[]>asList(
- new Object[] { TestExecutionMode.CLUSTER } );
+ public static void initialize() throws Exception {
+ Configuration config = new Configuration();
+ config.setInteger(ConfigConstants.LOCAL_NUMBER_TASK_MANAGER, NUM_TASK_MANAGERS);
+ config.setInteger(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS, NUM_SLOTS);
+ config.setInteger(ConfigConstants.TASK_MANAGER_MEMORY_SIZE_KEY, 12);
+ config.setBoolean(ConfigConstants.LOCAL_START_WEBSERVER, true);
+
+ File logDir = File.createTempFile("TestBaseUtils-logdir", null);
+ assertTrue("Unable to delete temp file", logDir.delete());
+ assertTrue("Unable to create temp directory", logDir.mkdir());
+ File logFile = new File(logDir, "jobmanager.log");
+ File outFile = new File(logDir, "jobmanager.out");
+
+ Files.createFile(logFile.toPath());
+ Files.createFile(outFile.toPath());
+
+ config.setString(ConfigConstants.JOB_MANAGER_WEB_LOG_PATH_KEY, logFile.getAbsolutePath());
+
+ cluster = new ForkableFlinkMiniCluster(config, false);
+ cluster.start();
+
+ port = cluster.webMonitor().get().getServerPort();
}
@Test
@@ -128,7 +136,8 @@ public class WebFrontendITCase extends MultipleProgramsTestBase {
JsonNode taskManager = taskManagers.get(0);
assertNotNull(taskManager);
- assertEquals(4, taskManager.get("freeSlots").asInt());
+ assertEquals(NUM_SLOTS, taskManager.get("slotsNumber").asInt());
+ assertTrue(taskManager.get("freeSlots").asInt() <= NUM_SLOTS);
}
catch(Exception e) {
e.printStackTrace();
@@ -137,22 +146,16 @@ public class WebFrontendITCase extends MultipleProgramsTestBase {
}
@Test
- public void getLogAndStdoutFiles() {
- try {
- WebMonitorUtils.LogFileLocation logFiles = WebMonitorUtils.LogFileLocation.find(cluster.configuration());
+ public void getLogAndStdoutFiles() throws Exception {
+ WebMonitorUtils.LogFileLocation logFiles = WebMonitorUtils.LogFileLocation.find(cluster.configuration());
- FileUtils.writeStringToFile(logFiles.logFile, "job manager log");
- String logs = getFromHTTP("http://localhost:" + port + "/jobmanager/log");
- assertTrue(logs.contains("job manager log"));
+ FileUtils.writeStringToFile(logFiles.logFile, "job manager log");
+ String logs = getFromHTTP("http://localhost:" + port + "/jobmanager/log");
+ assertTrue(logs.contains("job manager log"));
- FileUtils.writeStringToFile(logFiles.stdOutFile, "job manager out");
- logs = getFromHTTP("http://localhost:" + port + "/jobmanager/stdout");
- assertTrue(logs.contains("job manager out"));
- }
- catch (Exception e) {
- e.printStackTrace();
- fail(e.getMessage());
- }
+ FileUtils.writeStringToFile(logFiles.stdOutFile, "job manager out");
+ logs = getFromHTTP("http://localhost:" + port + "/jobmanager/stdout");
+ assertTrue(logs.contains("job manager out"));
}
@Test
@@ -161,7 +164,6 @@ public class WebFrontendITCase extends MultipleProgramsTestBase {
String config = getFromHTTP("http://localhost:" + port + "/jobmanager/config");
Map<String, String> conf = WebMonitorUtils.fromKeyValueJsonArray(config);
- assertTrue(conf.get(ConfigConstants.JOB_MANAGER_WEB_LOG_PATH_KEY).startsWith(logDir.toString()));
assertEquals(
cluster.configuration().getString("taskmanager.numberOfTaskSlots", null),
conf.get(ConfigConstants.TASK_MANAGER_NUM_TASK_SLOTS));
@@ -172,8 +174,11 @@ public class WebFrontendITCase extends MultipleProgramsTestBase {
}
}
- @Test(timeout = 15000)
+ @Test
public void testStop() throws Exception {
+ // this only works if there is no active job at this point
+ assertTrue(cluster.getCurrentlyRunningJobsJava().isEmpty());
+
// Create a task
final JobVertex sender = new JobVertex("Sender");
sender.setParallelism(2);
@@ -184,24 +189,34 @@ public class WebFrontendITCase extends MultipleProgramsTestBase {
cluster.submitJobDetached(jobGraph);
+ // wait for job to show up
+ while (cluster.getCurrentlyRunningJobsJava().isEmpty()) {
+ Thread.sleep(10);
+ }
+
final FiniteDuration testTimeout = new FiniteDuration(2, TimeUnit.MINUTES);
final Deadline deadline = testTimeout.fromNow();
- try (HttpTestClient client = new HttpTestClient("localhost", port)) {
- // Request the file from the web server
- client.sendDeleteRequest("/jobs/" + jid + "/stop", deadline.timeLeft());
- HttpTestClient.SimpleHttpResponse response = client.getNextResponse(deadline.timeLeft());
-
- assertEquals(HttpResponseStatus.OK, response.getStatus());
- assertEquals(response.getType(), MimeTypes.getMimeTypeForExtension("json"));
- assertEquals("{}", response.getContent());
+ while (!cluster.getCurrentlyRunningJobsJava().isEmpty()) {
+ try (HttpTestClient client = new HttpTestClient("localhost", port)) {
+ // Request the file from the web server
+ client.sendDeleteRequest("/jobs/" + jid + "/stop", deadline.timeLeft());
+ HttpTestClient.SimpleHttpResponse response = client.getNextResponse(deadline.timeLeft());
+
+ assertEquals(HttpResponseStatus.OK, response.getStatus());
+ assertEquals(response.getType(), MimeTypes.getMimeTypeForExtension("json"));
+ assertEquals("{}", response.getContent());
+ }
+
+ Thread.sleep(20);
}
-
- waitForTaskManagers();
}
- @Test(timeout = 15000)
+ @Test
public void testStopYarn() throws Exception {
+ // this only works if there is no active job at this point
+ assertTrue(cluster.getCurrentlyRunningJobsJava().isEmpty());
+
// Create a task
final JobVertex sender = new JobVertex("Sender");
sender.setParallelism(2);
@@ -212,34 +227,28 @@ public class WebFrontendITCase extends MultipleProgramsTestBase {
cluster.submitJobDetached(jobGraph);
- final FiniteDuration testTimeout = new FiniteDuration(2, TimeUnit.MINUTES);
- final Deadline deadline = testTimeout.fromNow();
-
- try (HttpTestClient client = new HttpTestClient("localhost", port)) {
- // Request the file from the web server
- client.sendGetRequest("/jobs/" + jid + "/yarn-stop", deadline.timeLeft());
-
- HttpTestClient.SimpleHttpResponse response = client
- .getNextResponse(deadline.timeLeft());
-
- assertEquals(HttpResponseStatus.OK, response.getStatus());
- assertEquals(response.getType(), MimeTypes.getMimeTypeForExtension("json"));
- assertEquals("{}", response.getContent());
+ // wait for job to show up
+ while (cluster.getCurrentlyRunningJobsJava().isEmpty()) {
+ Thread.sleep(10);
}
- waitForTaskManagers();
- }
-
- private void waitForTaskManagers() throws Exception {
- int count = 0;
+ final FiniteDuration testTimeout = new FiniteDuration(2, TimeUnit.MINUTES);
+ final Deadline deadline = testTimeout.fromNow();
- while (count != 4) {
- String json = getFromHTTP("http://localhost:" + port + "/taskmanagers/");
- JSONObject parsed = new JSONObject(json);
- JSONArray taskManagers = parsed.getJSONArray("taskmanagers");
- JSONObject taskManager = taskManagers.getJSONObject(0);
- count = taskManager.getInt("freeSlots");
+ while (!cluster.getCurrentlyRunningJobsJava().isEmpty()) {
+ try (HttpTestClient client = new HttpTestClient("localhost", port)) {
+ // Request the file from the web server
+ client.sendGetRequest("/jobs/" + jid + "/yarn-stop", deadline.timeLeft());
+
+ HttpTestClient.SimpleHttpResponse response = client
+ .getNextResponse(deadline.timeLeft());
+
+ assertEquals(HttpResponseStatus.OK, response.getStatus());
+ assertEquals(response.getType(), MimeTypes.getMimeTypeForExtension("json"));
+ assertEquals("{}", response.getContent());
+ }
+
+ Thread.sleep(20);
}
}
-
}
[3/4] flink git commit: [FLINK-3467] [runtime] Remove superfluous
objects from DataSourceTask.invoke
Posted by se...@apache.org.
[FLINK-3467] [runtime] Remove superfluous objects from DataSourceTask.invoke
This closes #1691
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/4870fd62
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/4870fd62
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/4870fd62
Branch: refs/heads/master
Commit: 4870fd62f0b8bfcc280aa326ffe948a119de6baf
Parents: e719eddb
Author: Greg Hogan <co...@greghogan.com>
Authored: Mon Feb 22 11:21:40 2016 -0500
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Mar 13 16:03:04 2016 +0100
----------------------------------------------------------------------
.../apache/flink/runtime/operators/DataSourceTask.java | 11 ++---------
1 file changed, 2 insertions(+), 9 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/4870fd62/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
----------------------------------------------------------------------
diff --git a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
index 960faf7..5eec40f 100644
--- a/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
+++ b/flink-runtime/src/main/java/org/apache/flink/runtime/operators/DataSourceTask.java
@@ -136,26 +136,19 @@ public class DataSourceTask<OT> extends AbstractInvokable {
final Collector<OT> output = this.output;
if (objectReuseEnabled) {
- OT reuse1 = serializer.createInstance();
- OT reuse2 = serializer.createInstance();
- OT reuse3 = serializer.createInstance();
+ OT reuse = serializer.createInstance();
// as long as there is data to read
while (!this.taskCanceled && !format.reachedEnd()) {
OT returned;
- if ((returned = format.nextRecord(reuse1)) != null) {
+ if ((returned = format.nextRecord(reuse)) != null) {
output.collect(returned);
-
- reuse1 = reuse2;
- reuse2 = reuse3;
- reuse3 = returned;
}
}
} else {
// as long as there is data to read
while (!this.taskCanceled && !format.reachedEnd()) {
-
OT returned;
if ((returned = format.nextRecord(serializer.createInstance())) != null) {
output.collect(returned);
[2/4] flink git commit: [FLINK-3590] [jdbc] Hide derby log in JDBC
tests
Posted by se...@apache.org.
[FLINK-3590] [jdbc] Hide derby log in JDBC tests
This closes #1773
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/2e78a36f
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/2e78a36f
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/2e78a36f
Branch: refs/heads/master
Commit: 2e78a36f97be5c448c69962a1b2151b4fd676406
Parents: acb051d
Author: zentol <ch...@apache.org>
Authored: Tue Mar 8 15:05:30 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Mar 13 16:03:04 2016 +0100
----------------------------------------------------------------------
.../flink/api/java/io/jdbc/DerbyUtil.java | 31 ++++++++++++++++++++
.../api/java/io/jdbc/JDBCInputFormatTest.java | 2 +-
.../api/java/io/jdbc/JDBCOutputFormatTest.java | 2 +-
.../api/java/io/jdbc/example/JDBCExample.java | 1 +
4 files changed, 34 insertions(+), 2 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/2e78a36f/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/DerbyUtil.java
----------------------------------------------------------------------
diff --git a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/DerbyUtil.java b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/DerbyUtil.java
new file mode 100644
index 0000000..523b8b5
--- /dev/null
+++ b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/DerbyUtil.java
@@ -0,0 +1,31 @@
+/**
+ * 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
+ * <p/>
+ * http://www.apache.org/licenses/LICENSE-2.0
+ * <p/>
+ * 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.api.java.io.jdbc;
+
+import java.io.OutputStream;
+
+@SuppressWarnings("unused")
+/**
+ * Utility class to disable derby logging
+ */
+public class DerbyUtil {
+ public static final OutputStream DEV_NULL = new OutputStream() {
+ public void write(int b) {
+ }
+ };
+}
http://git-wip-us.apache.org/repos/asf/flink/blob/2e78a36f/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormatTest.java b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormatTest.java
index b1d43df..3fb0278 100644
--- a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormatTest.java
+++ b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormatTest.java
@@ -57,7 +57,7 @@ public class JDBCInputFormatTest {
}
private static void prepareDerbyDatabase() throws ClassNotFoundException, SQLException {
- System.setProperty("derby.stream.error.field", "org.apache.flink.api.java.record.io.jdbc.DevNullLogStream.DEV_NULL");
+ System.setProperty("derby.stream.error.field", "org.apache.flink.api.java.io.jdbc.DerbyUtil.DEV_NULL");
String dbURL = "jdbc:derby:memory:ebookshop;create=true";
Class.forName("org.apache.derby.jdbc.EmbeddedDriver");
conn = DriverManager.getConnection(dbURL);
http://git-wip-us.apache.org/repos/asf/flink/blob/2e78a36f/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormatTest.java
----------------------------------------------------------------------
diff --git a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormatTest.java b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormatTest.java
index 276518b..1031b9a 100644
--- a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormatTest.java
+++ b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/JDBCOutputFormatTest.java
@@ -49,7 +49,7 @@ public class JDBCOutputFormatTest {
@BeforeClass
public static void setUpClass() throws SQLException {
try {
- System.setProperty("derby.stream.error.field", "org.apache.flink.api.java.record.io.jdbc.DevNullLogStream.DEV_NULL");
+ System.setProperty("derby.stream.error.field", "org.apache.flink.api.java.io.jdbc.DerbyUtil.DEV_NULL");
prepareDerbyDatabase();
} catch (ClassNotFoundException e) {
e.printStackTrace();
http://git-wip-us.apache.org/repos/asf/flink/blob/2e78a36f/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/example/JDBCExample.java
----------------------------------------------------------------------
diff --git a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/example/JDBCExample.java b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/example/JDBCExample.java
index 7b012ba..840a314 100644
--- a/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/example/JDBCExample.java
+++ b/flink-batch-connectors/flink-jdbc/src/test/java/org/apache/flink/api/java/io/jdbc/example/JDBCExample.java
@@ -57,6 +57,7 @@ public class JDBCExample {
}
private static void prepareTestDb() throws Exception {
+ System.setProperty("derby.stream.error.field", "org.apache.flink.api.java.io.jdbc.DerbyUtil.DEV_NULL");
String dbURL = "jdbc:derby:memory:ebookshop;create=true";
Class.forName("org.apache.derby.jdbc.EmbeddedDriver");
Connection conn = DriverManager.getConnection(dbURL);
[4/4] flink git commit: [FLINK-3472] [jdbc] Give a better exception
if jdbc column has a null value
Posted by se...@apache.org.
[FLINK-3472] [jdbc] Give a better exception if jdbc column has a null value
This closes #1772
Project: http://git-wip-us.apache.org/repos/asf/flink/repo
Commit: http://git-wip-us.apache.org/repos/asf/flink/commit/e719eddb
Tree: http://git-wip-us.apache.org/repos/asf/flink/tree/e719eddb
Diff: http://git-wip-us.apache.org/repos/asf/flink/diff/e719eddb
Branch: refs/heads/master
Commit: e719eddb9606351baccb45491f473563c2fe61fa
Parents: 2e78a36
Author: zentol <ch...@apache.org>
Authored: Wed Mar 9 11:11:31 2016 +0100
Committer: Stephan Ewen <se...@apache.org>
Committed: Sun Mar 13 16:03:04 2016 +0100
----------------------------------------------------------------------
.../flink/api/java/io/jdbc/JDBCInputFormat.java | 168 ++++++++++---------
1 file changed, 86 insertions(+), 82 deletions(-)
----------------------------------------------------------------------
http://git-wip-us.apache.org/repos/asf/flink/blob/e719eddb/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
----------------------------------------------------------------------
diff --git a/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java b/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
index 132edc4..b764350 100644
--- a/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
+++ b/flink-batch-connectors/flink-jdbc/src/main/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.java
@@ -191,89 +191,93 @@ public class JDBCInputFormat<OUT extends Tuple> extends RichInputFormat<OUT, Inp
*
* @param reuse Target Record.
*/
- private void addValue(OUT reuse) throws SQLException {
+ private void addValue(OUT reuse) throws IOException, SQLException {
for (int pos = 0; pos < columnTypes.length; pos++) {
- switch (columnTypes[pos]) {
- case java.sql.Types.NULL:
- reuse.setField(NullValue.getInstance(), pos);
- break;
- case java.sql.Types.BOOLEAN:
- reuse.setField(resultSet.getBoolean(pos + 1), pos);
- break;
- case java.sql.Types.BIT:
- reuse.setField(resultSet.getBoolean(pos + 1), pos);
- break;
- case java.sql.Types.CHAR:
- reuse.setField(resultSet.getString(pos + 1), pos);
- break;
- case java.sql.Types.NCHAR:
- reuse.setField(resultSet.getString(pos + 1), pos);
- break;
- case java.sql.Types.VARCHAR:
- reuse.setField(resultSet.getString(pos + 1), pos);
- break;
- case java.sql.Types.LONGVARCHAR:
- reuse.setField(resultSet.getString(pos + 1), pos);
- break;
- case java.sql.Types.LONGNVARCHAR:
- reuse.setField(resultSet.getString(pos + 1), pos);
- break;
- case java.sql.Types.TINYINT:
- reuse.setField(resultSet.getShort(pos + 1), pos);
- break;
- case java.sql.Types.SMALLINT:
- reuse.setField(resultSet.getShort(pos + 1), pos);
- break;
- case java.sql.Types.BIGINT:
- reuse.setField(resultSet.getLong(pos + 1), pos);
- break;
- case java.sql.Types.INTEGER:
- reuse.setField(resultSet.getInt(pos + 1), pos);
- break;
- case java.sql.Types.FLOAT:
- reuse.setField(resultSet.getDouble(pos + 1), pos);
- break;
- case java.sql.Types.REAL:
- reuse.setField(resultSet.getFloat(pos + 1), pos);
- break;
- case java.sql.Types.DOUBLE:
- reuse.setField(resultSet.getDouble(pos + 1), pos);
- break;
- case java.sql.Types.DECIMAL:
- reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
- break;
- case java.sql.Types.NUMERIC:
- reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
- break;
- case java.sql.Types.DATE:
- reuse.setField(resultSet.getDate(pos + 1).toString(), pos);
- break;
- case java.sql.Types.TIME:
- reuse.setField(resultSet.getTime(pos + 1).getTime(), pos);
- break;
- case java.sql.Types.TIMESTAMP:
- reuse.setField(resultSet.getTimestamp(pos + 1).toString(), pos);
- break;
- case java.sql.Types.SQLXML:
- reuse.setField(resultSet.getSQLXML(pos + 1).toString(), pos);
- break;
- default:
- throw new SQLException("Unsupported sql-type [" + columnTypes[pos] + "] on column [" + pos + "]");
-
- // case java.sql.Types.BINARY:
- // case java.sql.Types.VARBINARY:
- // case java.sql.Types.LONGVARBINARY:
- // case java.sql.Types.ARRAY:
- // case java.sql.Types.JAVA_OBJECT:
- // case java.sql.Types.BLOB:
- // case java.sql.Types.CLOB:
- // case java.sql.Types.NCLOB:
- // case java.sql.Types.DATALINK:
- // case java.sql.Types.DISTINCT:
- // case java.sql.Types.OTHER:
- // case java.sql.Types.REF:
- // case java.sql.Types.ROWID:
- // case java.sql.Types.STRUCT:
+ try {
+ switch (columnTypes[pos]) {
+ case java.sql.Types.NULL:
+ reuse.setField(NullValue.getInstance(), pos);
+ break;
+ case java.sql.Types.BOOLEAN:
+ reuse.setField(resultSet.getBoolean(pos + 1), pos);
+ break;
+ case java.sql.Types.BIT:
+ reuse.setField(resultSet.getBoolean(pos + 1), pos);
+ break;
+ case java.sql.Types.CHAR:
+ reuse.setField(resultSet.getString(pos + 1), pos);
+ break;
+ case java.sql.Types.NCHAR:
+ reuse.setField(resultSet.getString(pos + 1), pos);
+ break;
+ case java.sql.Types.VARCHAR:
+ reuse.setField(resultSet.getString(pos + 1), pos);
+ break;
+ case java.sql.Types.LONGVARCHAR:
+ reuse.setField(resultSet.getString(pos + 1), pos);
+ break;
+ case java.sql.Types.LONGNVARCHAR:
+ reuse.setField(resultSet.getString(pos + 1), pos);
+ break;
+ case java.sql.Types.TINYINT:
+ reuse.setField(resultSet.getShort(pos + 1), pos);
+ break;
+ case java.sql.Types.SMALLINT:
+ reuse.setField(resultSet.getShort(pos + 1), pos);
+ break;
+ case java.sql.Types.BIGINT:
+ reuse.setField(resultSet.getLong(pos + 1), pos);
+ break;
+ case java.sql.Types.INTEGER:
+ reuse.setField(resultSet.getInt(pos + 1), pos);
+ break;
+ case java.sql.Types.FLOAT:
+ reuse.setField(resultSet.getDouble(pos + 1), pos);
+ break;
+ case java.sql.Types.REAL:
+ reuse.setField(resultSet.getFloat(pos + 1), pos);
+ break;
+ case java.sql.Types.DOUBLE:
+ reuse.setField(resultSet.getDouble(pos + 1), pos);
+ break;
+ case java.sql.Types.DECIMAL:
+ reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
+ break;
+ case java.sql.Types.NUMERIC:
+ reuse.setField(resultSet.getBigDecimal(pos + 1).doubleValue(), pos);
+ break;
+ case java.sql.Types.DATE:
+ reuse.setField(resultSet.getDate(pos + 1).toString(), pos);
+ break;
+ case java.sql.Types.TIME:
+ reuse.setField(resultSet.getTime(pos + 1).getTime(), pos);
+ break;
+ case java.sql.Types.TIMESTAMP:
+ reuse.setField(resultSet.getTimestamp(pos + 1).toString(), pos);
+ break;
+ case java.sql.Types.SQLXML:
+ reuse.setField(resultSet.getSQLXML(pos + 1).toString(), pos);
+ break;
+ default:
+ throw new SQLException("Unsupported sql-type [" + columnTypes[pos] + "] on column [" + pos + "]");
+
+ // case java.sql.Types.BINARY:
+ // case java.sql.Types.VARBINARY:
+ // case java.sql.Types.LONGVARBINARY:
+ // case java.sql.Types.ARRAY:
+ // case java.sql.Types.JAVA_OBJECT:
+ // case java.sql.Types.BLOB:
+ // case java.sql.Types.CLOB:
+ // case java.sql.Types.NCLOB:
+ // case java.sql.Types.DATALINK:
+ // case java.sql.Types.DISTINCT:
+ // case java.sql.Types.OTHER:
+ // case java.sql.Types.REF:
+ // case java.sql.Types.ROWID:
+ // case java.sql.Types.STRUCT:
+ }
+ } catch (NullPointerException npe) {
+ throw new IOException("Encountered null value for column " + pos + ". Decimal, Numeric, Date, Time, Timestamp and SQLXML columns may not contain NULL values.");
}
}
}