title: Spark constantly resetting offset to X
body: <p>I'm developing spark structured streaming application with kafka. It's working fine except one thing. The problem is spark continuously resetting all partition's offset to X. It's consuming much network IO and CPU. If I add more kafka consumers CPU consumption is noticeable. Almost 25% of CPU usage while idle.
  Is it normal behaviour? Or I missing some configuration?</p>
  
  <p>I created minimal spark kafka consumer application for demonstration.</p>
  
  <p>This is logs that completely new startup and while idling.</p>
  
  <pre><code>19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:29 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  19/12/11 16:24:30 INFO Fetcher: [Consumer clientId=consumer-1, groupId=spark-kafka-source-b2457938-d427-47e2-b90a-7c6f0d85904b--1563005380-driver-0] Resetting offset for partition testJson-0 to offset 2.
  
  </code></pre>
  
  <p>Project</p>
  
  <pre><code>import org.apache.spark.SparkConf;
  import org.apache.spark.api.java.JavaSparkContext;
  import org.apache.spark.sql.Dataset;
  import org.apache.spark.sql.Row;
  import org.apache.spark.sql.SparkSession;
  import org.apache.spark.sql.streaming.OutputMode;
  import org.apache.spark.sql.streaming.StreamingQuery;
  import org.apache.spark.sql.streaming.StreamingQueryException;
  import org.apache.spark.sql.types.DataTypes;
  import org.apache.spark.sql.types.StructField;
  import org.apache.spark.sql.types.StructType;
  
  import static org.apache.spark.sql.functions.*;
  
  public class Main {
  
      public Main() throws StreamingQueryException {
          StructType schema = DataTypes.createStructType(new StructField[]{
                  DataTypes.createStructField("id", DataTypes.LongType, false),
                  DataTypes.createStructField("name", DataTypes.StringType, false),
                  DataTypes.createStructField("value", DataTypes.LongType, false),
          });
  
          SparkConf conf =  new SparkConf(true)
                  .setMaster("local[1]")
                  .set("spark.default.parallelism", "1")
                  .setAppName("spark-kafka-demo1");
  
          JavaSparkContext context = new JavaSparkContext(conf);
  
          SparkSession session = SparkSession
                  .builder()
                  .config(conf)
                  .sparkContext(context.sc())
                  .appName("spark-kafka-demo1")
                  .getOrCreate();
  
          Dataset&lt;Row&gt; dataset = session
                  .readStream()
                  .format("kafka")
                  .option("kafka.bootstrap.servers", "192.168.0.201:9092")
                  .option("subscribe", "testJson")
                  .load()
                  .selectExpr("CAST(value AS STRING) as message")
                  .select(from_json(col("message"), schema).as("t"));
  
          StreamingQuery query = dataset
              .groupBy("t.name")
                  .agg(sum("t.value"))
              .writeStream()
                  .format("console")
                  .option("truncate", false)
                  .outputMode(OutputMode.Update())
                  .start();
  
          query.awaitTermination();
      }
  
      public static void main(String[] args) throws StreamingQueryException {
          new Main();
      }
  }
  
  </code></pre>
  
  <p>Maven dependencies</p>
  
  <pre><code>        &lt;dependency&gt;
              &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
              &lt;artifactId&gt;spark-sql_2.11&lt;/artifactId&gt;
              &lt;version&gt;2.4.4&lt;/version&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
              &lt;artifactId&gt;spark-streaming_2.11&lt;/artifactId&gt;
              &lt;version&gt;2.4.4&lt;/version&gt;
              &lt;scope&gt;provided&lt;/scope&gt;
          &lt;/dependency&gt;
          &lt;dependency&gt;
              &lt;groupId&gt;org.apache.spark&lt;/groupId&gt;
              &lt;artifactId&gt;spark-sql-kafka-0-10_2.11&lt;/artifactId&gt;
              &lt;version&gt;2.4.4&lt;/version&gt;
          &lt;/dependency&gt;
  </code></pre>