Using JDBCInputFormat with variable tuple size (apache-flink)
<p>I want to write a generic flink job in java, which can take any SQL-SELECT query, run it against a SQL-database and write it into a Elasticsearch index.</p>

<p>One of the problems I have to solve is creating a DataSource for a JDBC-Connection. I want to use the <a href="https://ci.apache.org/projects/flink/flink-docs-release-0.10/api/java/org/apache/flink/api/java/io/jdbc/JDBCInputFormat.html" rel="nofollow">JDBCInputFormat</a>. I followed the example in the <a href="https://ci.apache.org/projects/flink/flink-docs-release-0.10/apis/programming_guide.html#data-sources" rel="nofollow">documentation data source</a>.</p>

<p>The problem is, the generic type <code>DataSource</code> type must be specified. And I can only use a <code>Tuple</code> type, because <code>JDBCInputFormat</code> generic type <code>OUT</code> extends <code>Tuple</code>. But I do not know at compile time which <code>Tuple</code> I will use.</p>

<ol>
<li>Do I interpret something wrong?</li>
<li>Is there another jdbc <code>InputFormat</code> I can use?</li>
<li>Is there a way to specify <code>Tuple</code> as a generic type?</li>
</ol>

<p>I use java 7 and apache-flink 0.10.2</p>

<p>I tried to use <code>Tuple25</code> with only Strings in it, but I get an exception.</p>

<p>Here follows code and then the exception.</p>

<pre><code>DataSource&lt;StringsTuple25&gt; database = flink.createInput(
JDBCInputFormat.buildJDBCInputFormat()//
  .setDrivername(getDatabaseDriverName())//
  .setDBUrl(getDatabaseUrl())//
  .setUsername(getDatabaseUsername())//
  .setPassword(getDatabasePassword())//
  .setQuery(getQuery())//
  .finish(), 
  StringsTuple25.typeInformation()
);
</code></pre>

<p>My <code>StringTuple25</code> class</p>

<pre><code>public class StringsTuple25 extends
  Tuple25&lt;String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String, String&gt; {

  private static final long serialVersionUID = 1L;

  public static TypeInformation&lt;?&gt; typeInformation() {
    TypeInformation&lt;String&gt;[] types = new TypeInformation[25];
    Arrays.fill(types, STRING_TYPE_INFO);
    return new TupleTypeInfo&lt;&gt;(Tuple25.class,types);
  }
}
</code></pre>

<p>And I get this exception:</p>

<pre><code>Caused by: java.io.IOException: Tuple size does not match columncount
  at org.apache.flink.api.java.io.jdbc.JDBCInputFormat.extractTypes(JDBCInputFormat.java:180)
  at org.apache.flink.api.java.io.jdbc.JDBCInputFormat.nextRecord(JDBCInputFormat.java:162)
  at org.apache.flink.api.java.io.jdbc.JDBCInputFormat.nextRecord(JDBCInputFormat.java:51)
  at org.apache.flink.runtime.operators.DataSourceTask.invoke(DataSourceTask.java:169)
  at org.apache.flink.runtime.taskmanager.Task.run(Task.java:584)
  at java.lang.Thread.run(Thread.java:745)
</code></pre>