title: KafkaUtils API | offset management | Spark Streaming
body: <p>I am trying to manage kafka offsets for exactly once semantics.</p>
  
  <p>Facing problem while creating a direct stream using offset map as follows :</p>
  
  <pre><code>val fromOffsets : (TopicAndPartition, Long) = TopicAndPartition(metrics_rs.getString(1), metrics_rs.getInt(2)) -&gt; metrics_rs.getLong(3)
  
  KafkaUtils.createDirectStream[String, String, StringDecoder, StringDecoder,(String, String)] (ssc,kafkaParams,fromOffsets,messageHandler)
  </code></pre>
  
  <p>here,</p>
  
  <pre><code>val messageHandler =
        (mmd: MessageAndMetadata[String, String]) =&gt; mmd.message.length
  </code></pre>
  
  <p>And </p>
  
  <pre><code>metrics_rs = metricsStatement.executeQuery("SELECT part,off from metrics.txn_offsets where topic='"+t+''' )
  </code></pre>
  
  <p>I guess I am doing something wrong with the declaration style...if you could help.
  The compilation error says "too many type arguments for createDirectStream"</p>