ignite-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "Andrey Aleksandrov (Jira)" <j...@apache.org>
Subject [jira] [Updated] (IGNITE-11724) IgniteSpark integration forget to close the IgniteContext and stops the client node in case if error during PairFunction logic
Date Mon, 21 Oct 2019 11:17:00 GMT

     [ https://issues.apache.org/jira/browse/IGNITE-11724?page=com.atlassian.jira.plugin.system.issuetabpanels:all-tabpanel
]

Andrey Aleksandrov updated IGNITE-11724:
----------------------------------------
    Description: 
Next code could hang in case if PairFunction logic will throw the exception:
{code:java}
public class Example {
    public static void main(String[] args) {
        String configPath = "/home/andrei/BDP/big-data-accelerator/modules/gridgain-spark-loader-examples/config/client.xml";
        IgniteSparkSession igniteSession = IgniteSparkSession.builder()
                .appName("Spark Ignite catalog example")
                .master("local")
                .config("ignite.disableSparkSQLOptimization", true)
                .igniteConfig(configPath)
                .getOrCreate();
        JavaSparkContext sparkCtx = new JavaSparkContext(igniteSession.sparkContext());
        final JavaRDD<Row> records = sparkCtx.parallelize(Arrays.asList(
                new GenericRow()
        ));
        JavaPairRDD<Integer, Integer> rdd_records = records.mapToPair(new PairFunction<Row,
Integer, Integer>() {
            @Override
            public Tuple2<Integer, Integer> call(Row row) throws Exception {
                throw new IllegalStateException("some error");
            }
        });
        JavaIgniteContext<Integer, Integer> igniteContext = new JavaIgniteContext<>(sparkCtx,
configPath);
        JavaIgniteRDD<Integer, Integer> igniteRdd = igniteContext.<Integer, Integer>fromCache("Person");
        igniteRdd.savePairs(rdd_records);
        igniteContext.close(true);
    }
}

Looks like next internal code (saveValues method)should also close the IgniteContext in case
of an unexpected exception, not only data streamer:

try {
     it.foreach(value ⇒

{          val key = affinityKeyFunc(value, node.orNull)           streamer.addData(key,
value)        }

)
     }
     finally

{         streamer.close()     }

})
 }

{code}

  was:
Next code could hang in case if PairFunction logic will throw the exception:

{code:java}

JavaPairRDD<Key, Value> rdd_records = records.mapToPair(new MapFunction());

JavaIgniteContext<Key, Value> igniteContext = new JavaIgniteContext<>(sparkCtx,
configUrl);

JavaIgniteRDD<Key, Value> igniteRdd = igniteContext.<Key, Value>fromCache(cacheName);

igniteRdd.savePairs(rdd_records);

Looks like next internal code (saveValues method)should also close the IgniteContext in case
of an unexpected exception, not only data streamer:

try {
     it.foreach(value ⇒

{          val key = affinityKeyFunc(value, node.orNull)           streamer.addData(key,
value)        }

)
     }
     finally

{         streamer.close()     }

})
 }

{code}


> IgniteSpark integration forget to close the IgniteContext and stops the client node in
case if error during PairFunction logic 
> -------------------------------------------------------------------------------------------------------------------------------
>
>                 Key: IGNITE-11724
>                 URL: https://issues.apache.org/jira/browse/IGNITE-11724
>             Project: Ignite
>          Issue Type: Bug
>          Components: spark
>    Affects Versions: 2.8
>            Reporter: Andrey Aleksandrov
>            Assignee: Alexey Zinoviev
>            Priority: Major
>              Labels: await
>             Fix For: 2.8
>
>          Time Spent: 20m
>  Remaining Estimate: 0h
>
> Next code could hang in case if PairFunction logic will throw the exception:
> {code:java}
> public class Example {
>     public static void main(String[] args) {
>         String configPath = "/home/andrei/BDP/big-data-accelerator/modules/gridgain-spark-loader-examples/config/client.xml";
>         IgniteSparkSession igniteSession = IgniteSparkSession.builder()
>                 .appName("Spark Ignite catalog example")
>                 .master("local")
>                 .config("ignite.disableSparkSQLOptimization", true)
>                 .igniteConfig(configPath)
>                 .getOrCreate();
>         JavaSparkContext sparkCtx = new JavaSparkContext(igniteSession.sparkContext());
>         final JavaRDD<Row> records = sparkCtx.parallelize(Arrays.asList(
>                 new GenericRow()
>         ));
>         JavaPairRDD<Integer, Integer> rdd_records = records.mapToPair(new PairFunction<Row,
Integer, Integer>() {
>             @Override
>             public Tuple2<Integer, Integer> call(Row row) throws Exception {
>                 throw new IllegalStateException("some error");
>             }
>         });
>         JavaIgniteContext<Integer, Integer> igniteContext = new JavaIgniteContext<>(sparkCtx,
configPath);
>         JavaIgniteRDD<Integer, Integer> igniteRdd = igniteContext.<Integer,
Integer>fromCache("Person");
>         igniteRdd.savePairs(rdd_records);
>         igniteContext.close(true);
>     }
> }
> Looks like next internal code (saveValues method)should also close the IgniteContext
in case of an unexpected exception, not only data streamer:
> try {
>      it.foreach(value ⇒
> {          val key = affinityKeyFunc(value, node.orNull)           streamer.addData(key,
value)        }
> )
>      }
>      finally
> {         streamer.close()     }
> })
>  }
> {code}



--
This message was sent by Atlassian Jira
(v8.3.4#803005)

Mime
View raw message