Mocked Kinesis (Localstack) with PySpark Streaming

Continuing with the same PySpark (ver 2.1.0, Python3.5, etc.) setup explained in an earlier post. In order to connect to the mocked Kinesis stream on Localstack from PySpark use the kinesis_wordcount_asl.py script located in Spark external/ (connector/) folder.

(a) Update value of master in kinesis_wordcount_asl.py

Update value of master(local[n], spark://localhost:7077, etc) in SparkContext in kinesis_wordcount_asl.py:
    sc = SparkContext(appName="PythonStreamingKinesisWordCountAsl",master="local[2]")

(b) Add aSpark compiled jars to Spark Driver/ Executor Classpath

As explained in step (III) of an earlier post, to work with Localstack a few changes were done to the KinesisReceiver.scala onStart() to explicitly set endPoint on kinesis, dynamoDb, cloudWatch clients. Accordingly the compiled aSpark jars with the modifications need to be added to Spark Driver/ Executor classpath.

• For Spark local mode (master="local[n]"): additions to classpath can be exported in the SPARK_CLASSPATH variable.

     export aSPARK_PROJ_HOME="/Downlaod/Location/aSpark"
    export SPARK_CLASSPATH="${aSPARK_PROJ_HOME}/target/original-aSpark_1.0-2.1.0.jar:${aSPARK_PROJ_HOME}/target/scala-2.11/classes:${aSPARK_PROJ_HOME}/target/scala-2.11/jars/*"

•  For Spark Standalone mode: "spark.executor.extraClassPath" needs to be set in either spark-defaults.conf or added as a SparkConf to SparkContext (see (II)(a))

(c) Ensure SPARK_HOME, PYSPARK_PYTHON & PYTHONPATH variables are exported.

(d) Run kinesis_wordcount_asl

    python3.5 ${SPARK_HOME}/external/kinesis-asl/src/main/python/examples/streaming/kinesis_wordcount_asl.py SampleKinesisApplication myFirstStream http://localhost:4566/ us-east-1

• put-records to Localstack Kinesis

    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata abcd"

• Count of the words streamed (put) will show up on the kinesis_wordcount_asl console

 

Spark Streaming with Kinesis mocked on Localstack

In this post we get a Spark streaming application working with AWS Kinesis stream, a mocked version of Kinesis running locally on Localstack. In earlier posts we have explained how to get Localstack running and various AWS services up on Localstack. The client connections to AWS services (Localstack) is done using AWS cli and AWS Java-Sdk v1.

Environment: This set-up continues on a Ubuntu20.04, with Java-8, Maven-3.6x, Docker-24.0x, Python3.5, PySpark/ Spark-2.1.0, Localstack-3.8.1, AWS Java-Sdk-v1 (ver.1.12.778),

Once the Localstack installation is done, steps to follow are:

(I) Start Localstack
    # Start locally
    localstack start

    That should get Localstack should be running on: http://localhost:4566

(II) Check Kinesis services from CLI on Localstack

    # List Streams
    aws --endpoint-url=http://localhost:4566 kinesis list-streams

    # Create Stream
    aws --endpoint-url=http://localhost:4566 kinesis create-stream --stream-name myFirstStream --shard-count 1

    # List Streams
    aws --endpoint-url=http://localhost:4566 kinesis list-streams

    # describe-stream-summary
    aws --endpoint-url=http://localhost:4566 kinesis describe-stream-summary --stream-name myFirstStream

    # Put Record
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata abcd"
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata efgh"

(III) Connect to Kinesis from Spark Streaming

• Download & Build a sample aSpark - Java Kinesis application.
• The code is similar to Spark's kinesis-asl from external (connector) module. Except for a few changes to KinesisReceiver.scala onStart() method to explicitly set endPoint on kinesis, dynamoDb, cloudWatch clients. This enables Localstack endPoint url to be plugged into kinesis, dynamoDb & cloudwatch.

    # Build
    mvn install -DskipTests=true -Dcheckstyle.skip

    # Run JavaKinesisWordCountASL with Localstack

• JavaKinesisWordCountASL SampleKinesisApplication myFirstStream http://localhost:4566/
  

• runJavaKinesisWordCountASL.sh script located in sbin/ folder of the aSpark project can be used to run JavaKinesisWordCoundASL from the shell
  

(IV) Add Data to Localstack Kinesis & View Counts on Console
    a) Put Record from cli
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata abcd"
    aws  --endpoint-url=http://localhost:4566 kinesis put-record --stream-name myFirstStream --partition-key 123 --data "testdata efgh"

    b) Alternatively Put records from Java Kinesis application
    Download, build & run AmazonKinesisRecordProducerSample.java
    
    c) Now check the output console of JavaKinesisWordCountASL run in step (III) above. Counts of the words streamed from Localstack Kinesis will be displayed on the console.

Debugging Spark Scala/ Java components

In continuation to the earlier post regarding debugging Pyspark, here we show how to debug the Spark Scala/ Java side. Spark is a distributed processing environment and has Scala Api's for connecting from different languages like Python & Java. The high level Pyspark Architecture is shown here.

For debugging the Spark Scala/ Java components as these run within the JVM, it's easy to make use of Java Tooling Options for remote debugging from any compatible IDE such as Idea (Eclipse longer supports Scala). A few points to remember:

• Multiple JVMs in Spark: Since Spark is a distributed application, it involves several components like the Master/ Driver, Slave/ Worker, Executor. In a real world truly distributed setting, each of the components runs in its own separate JVM on separated Physical machines. So be clear about which component you are exactly wanting to debug & set up the Tooling options accordingly targetting the specific JVM instance.

• Two-way connectivity between IDE & JVM: At the same time there should be a two-way network connectivity between the IDE (debugger) & the running JVM instance

• Debugging Locally: Debugging is mostly a dev stage activity & done locally. So it may be better to debug on a a Spark cluster running locally. This could be either on a Spark Spark cluster or a Spark run locally (master=local[n]/ local[*]).

Steps:

Environment: Ubuntu-20.04 having Java-8, Spark/Pyspark (ver 2.1.0), Python3.5, Idea-Intelli (ver 2024.3), Maven3.6

(I) Idea Remote JVM Debugger
In Idea > Run/ Debug Config > Edit > Remote JVM Debug.

• Start Debugger in Listen to Remote JVM Mode
• Enable Auto Restart

(II)(a) Debug Spark Standlone cluster
Key features of the Spark Standalone cluster are:

• Separate JVMs for Master, Slave/ Worker, Executor
• All could run on a single dev box, provided enough resources (Mem, CPU) are available
• Scripts inside SPARK_HOME/sbin folder like start-master.sh, start-slave.sh (start-worker.sh), etc to start the services

In order to Debug lets say some Executor, a Spark Standalone cluster could be started off with 1 Master, 1 Worker, 1 Executor.   

    # Start Master (Check http://localhost:8080/ to get Master URL/ PORT)
    ./sbin/start-master.sh 

    # Start Slave/ Worker
    ./sbin/start-slave.sh spark://MASTER_URL:<MASTER_PORT>

    # Add Jvm tooling to extraJavaOption to spark-defaults.conf
    spark.executor.extraJavaOptions  -agentlib:jdwp=transport=dt_socket,server=n,address=localhost:5005,suspend=n

    # The value could instead be passed as a conf to SparkContext in Python script:
    from pyspark.conf import SparkConf
    confVals = SparkConf()
    confVals.set("spark.executor.extraJavaOptions","-agentlib:jdwp=transport=dt_socket,server=n,address=localhost:5005,suspend=y")
    sc = SparkContext(master="spark://localhost:7077",appName="PythonStreamingStatefulNetworkWordCount1",conf=confVals)

(II)(b) Debug locally with master="local[n]"

• In this case a local Spark cluster is spun up via scripts like spark-shell, spark-submit, etc. located inside the bin/ folder
• The different components Master, Worker, Executor all run within one JVM as threads, where the value n is the no of threads, (set n=2)
• Export JAVA_TOOL_OPTIONS before in the terminal from which the Pyspark script will be run

        export JAVA_TOOL_OPTIONS="-agentlib:jdwp=transport=dt_socket,server=n,suspend=n,address=5005"

(III) Execute PySpark Python script
    python3.5 ${SPARK_HOME}/examples/src/main/python/streaming/network_wordcount.py localhost 9999

This should start off the Pyspark & connect the Executor JVM to the waiting Idea Remote debugger instance for debugging.

Pyspark in Eclipse with PyDev

This post captures the steps to get Spark (ver 2.1) working within Eclipse (ver 2024-06 (4.32)) using the PyDev (ver 12.1) plugin. The OS is Ubuntu-20.04 with Java-8, Python 3.x & Maven 3.6.

(I) Compile Spark code

The Spark code is downloaded & compiled from a location "SPARK_HOME".

    export SPARK_HOME="/SPARK/DOWNLOAD/LOCATION"

    cd ${SPARK_HOME}

    mvn install -DskipTests=true -Dcheckstyle.skip -o

(Issue: For a "Failed to execute goal org.scalastyle:scalastyle-maven-plugin:0.8.0:check":
Copy scalastyle-config.xml to the sub-project (next to pom.xml) having the error.

(II) Compile Pyspark
    (a) Install Pyspark dependencies

• Install Pandoc

        sudo apt-get install pandoc

• Install a compatible older Pypandoc (ver 1.5)

        pip3 install pypandoc==1.5

• Install a compatible older Python 3.x (ver 3.5)

        sudo add-apt-repository ppa:deadsnakes/ppa

        sudo apt-get install python3.5
    
    (b) Build Pyspark

        cd ${SPARK_HOME}/python

        export PYSPARK_PYTHON=python3.5

        # Build - creates ${SPARK_HOME}/python/build
        python3.5 setup.py

        # Dist - creates ${SPARK_HOME}/python/dist
        python3.5 setup.py sdist

    (c) export PYTHON_PATH

    export PYTHONPATH=$PYTHONPATH:${SPARK_HOME}/python/:${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip:${SPARK_HOME}/python/pyspark/shell.py;
    
(III) Run Pyspark from console
    Pyspark setup is done & stanalone examples code should run. Ensure variables ${SPARK_HOME}, ${PYSPARK_PYTHON} & ${PYTHONPATH} are all correctly exported (steps (I), (II)(b) & (II)(c) above):

    python3.5 ${SPARK_HOME} /python/build/lib/pyspark/examples/src/main/python/streaming/network_wordcount.py localhost 9999

(IV) Run Pyspark on PyDev in Eclipse

    (a) Eclipse with PyDev plugin installed:
    Set-up tested on Eclipse (ver 2024-06 (4.32.0)) and PyDev plugin (ver 12.1x).
 
    (b) Import the spark project in Eclipse
    There would be compilation errors due to missing Spark Scala classes.

    (c) Add Target jars for Spark Scala classes
    Eclipse no longer has support for Scala so the corresponding Spark Scala classes are missing. A work around is to add the Scala target jars compiled using mvn (in step (I) above) manually to: 

        spark-example > Properties > Java Build Path > Libraries
   

    (d) Add PyDev Interpreter for Python3.5
    Go to: spark-example > Properties > PyDev - Interpreter/ Grammar > Click to confure an Interpreter not listed > Open Interpreter Preferences Page > New > Choose from List:  

    & Select /usr/bin/python3.5

 

 On the same page, under the Environment tab add a variable named "PYSPARK_PYTHON" having value "python3.5"

    (e) Set up PYTHONPATH for PyDev

    spark-example > Properties > PyDev - PYTHONPATH

• Under String Substitution Variables add a variable with name "SPARK_HOME" & value "/SPARK/DOWNLOAD/LOCATION" (same location added in Step (I)). 

 

• Under External Libraries, Choose Add based on variable, add 3 entries:
  

                 ${SPARK_HOME}/python/

                 ${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip

                 ${SPARK_HOME}/python/lib/py4j-0.10.4-src.zip

   With that Pyspark should be properly set-up within PyDev.

    (f) Run Pyspark from Eclipse

    Right click on network_wordcount.py > Run as > Python run
    (You can further change Run Configurations > Arguments & provide program arguments, e.g. "localhost 9999")

Scala IDE no more

Sad that Scala IDE for Eclipse is no longer supported. While it was a great to have Scala integrated within Eclipse, guess the headwinds were too strong!

Working with Moto & Lambci Lambda Docker Images

Next up on Mock for clouds is Moto. Moto is primarily for running tests within the Python ecosystem.

Moto does offer a standalone server mode for a other langauges. General sense was that the standalone Moto server would offer the AWS services which will be accessible from the cli & non-Python SDKs. Gave Moto a shot with the same AWS services tried with Localstack.

(I) Set-up

While installing Moto ran into a couple of dependency conflicts across moto, boto3, botocore, requests, s3transfer & in turn with the installed awscli. With some effort reached a sort of dynamic equillibrium with (installed via pip):

• awscli                       1.36.11             
• boto3                        1.35.63             
• botocore                   1.35.70             
• moto                         5.0.21              
• requests                   2.32.2                          
• s3transfer                0.10.4  

(II) Start Moto Server

    # Start Moto
    moto_server -p3000

    # Start Moto as Docker (Sticking to this option)
    docker run --rm -p 5000:5000 --name moto motoserver/moto:latest

(III) Invoke services on Moto

    (a) S3
    # Create bucket
    aws --endpoint-url=http://localhost:5000 s3 mb s3://test-buck

    # Copy item to bucket
    aws --endpoint-url=http://localhost:5000 s3 cp a1.txt s3://test-buck

    # List bucket
    aws --endpoint-url=http://localhost:5000 s3 ls s3://test-buck

--
    (b) SQS
    # Create queue
    aws --endpoint-url=http://localhost:5000 sqs create-queue --queue-name test-q

    # List queues
    aws --endpoint-url=http://localhost:5000 sqs list-queues

    # Get queue attribute
    aws --endpoint-url=http://localhost:5000 sqs get-queue-attributes --queue-url http://localhost:5000/123456789012/test-q --attribute-names All

--
    (c) IAM
    ## Issue: Moto does a basic check of user role & gives an AccessDeniedException when calling Lambda CreateFunction operation
    ## So have to create a specific IAM role (https://github.com/getmoto/moto/issues/3944#issuecomment-845144036) in Moto for the purpose.

    aws iam --region=us-east-1 --endpoint-url=http://localhost:5000 create-role --role-name "lambda-test-role" --assume-role-policy-document "some policy" --path "/lambda-test/"

--
    (d) Lambda
    # Create Java function
    aws --endpoint-url=http://localhost:5000 lambda create-function --function-name test-j-div --zip-file fileb://original-java-basic-1.0-SNAPSHOT.jar --handler example.HandlerDivide::handleRequest --runtime java8.al2 --role arn:aws:iam::123456789012:role/lambda-test/lambda-test-role

    # List functions
    aws --endpoint-url=http://localhost:5000 lambda list-functions

    # Invoke function (Fails!)
    aws --endpoint-url=http://localhost:5000 lambda invoke --function-name test-j-div --payload '[235241,17]' outputJ.txt

    The invoke function fails with the message:
    "WARNING - Unable to parse Docker API response. Defaulting to 'host.docker.internal'
    <class 'json.decoder.JSONDecodeError'>::Expecting value: line 1 column 1 (char 0)
    error running docker: Error while fetching server API version: ('Connection aborted.', FileNotFoundError(2, 'No such file or directory'))".
    
    Retried this from AWS Java-SDK & for other nodejs & python function but nothing worked. While this remains unsolved for now, check out Lambci docker option next.

(IV) Invoke services on Lambci Lambda Docker Images:

    Moto Lambda docs also mention its dependent docker images from the lambci/lambda & mlupin/docker-lambda (for new ones). Started off with a slightly older java8.al2 docker image from lambci/lambda.

    # Download lambci/lambda:java8.al2
    docker pull lambci/lambda:java8.al2
    
    # Run lambci/lambda:java8.al2.   
    ## Ensure to run from the location which has the unzipped (unjarred) Java code
    ## Here it's run from a folder called data_dir_java which has the unzipped (unjarred) class file folders: com/, example/, META-INF/, net/ 

    docker run -e DOCKER_LAMBDA_STAY_OPEN=1 -p 9001:9001 -v "$PWD":/var/task:ro,delegated --name lambcijava8al2 lambci/lambda:java8.al2 example.HandlerDivide::handleRequest

    # Invoke Lambda
    aws --endpoint-url=http://localhost:9001 lambda invoke --function-name test-j-div --payload '[235241,17]' outputJ.txt

    This works!
 

AWS Lambda on Localstack using Java-SdK-v1

Continuing with Localstack, next is a closer look into the code to deploy and execute AWS Lambda code on Localstack from AWS Java-Sdk-v1. The localstack-lambda-java-sdk-v1 code uses the same structure used in localstack-aws-sdk-examples & fills in for the missing AWS Lambda bit.

The LambdaService class has 3 primary methods - listFunctions(), createFunction() & invokeFunction().  The static AWSLambda client is setup with Mock credentials and pointing to the Localstack endpoint.
 
The main() method first creates the function (createFunction()), if it does not exist.

• It builds a CreateFunctionRequest object with the handler, runtime, role, etc specified
• It also reads the jar file of the Java executable from the resources folder into a FunctionCode object & adds it to the CreateFunctionRequest
• Next a call is made to the AWSLambda client createFunction() with the CreateFunctionRequest which hits the running Localstack instance (Localstack set-up explained earlier).

If all goes well, control returns to main() which invokes the listFunctions() to show details of the created Lambda function (& all others functions existing).

Finally, there is call from main() to invokeFunction() method.

• Which invokes the recently created function with a InvokeRequest object filled with some test values as the payload.
• The response from the invoked function is a InvokeResult object who's payload contains the results of the lambda function computation.

Comments welcome, localstack-lambda-java-sdk-v1 is available to play around!

 

Mutable Argument Capture with Mockito

There are well known scenarios like caching, pooling, etc wherein object reuse is common. Testing these cases using a framework like Mockito could run into problems. Esp if there's a need to verify the arguments sent by the Caller of a Service, where the Service is mocked.

ArgumentCaptor (mockito) fails because it keeps references to the argument obj, which due to reuse by the caller only have the last/ latest updated value.    
The discussion here led to using Void Answer as one possible way to solve the issue. The following (junit-3+, mockito-1.8+, commons-lang-2.5) code explains the details.

1. Service: 

public class Service {

public void serve(MutableInt value) {

System.out.println("Service.serve(): "+value);

}

....

 

2. Caller:

public class Caller {

public void callService(Service service) {

MutableInt value = new MutableInt();

value.setValue(1);

service.serve(value);

value.setValue(2);

service.serve(value);

}

...

 

3.Tests:

public class MutableArgsTest extends TestCase{

List<MutableInt> multiValuesWritten;

@Mock

Service service;

 

/**

* Failure with ArgumentCaptor

*/

public void testMutableArgsWithArgCaptorFail() {

Caller caller = new Caller();

ArgumentCaptor<MutableInt> valueCaptor = 

ArgumentCaptor.forClass(MutableInt.class);

caller.callService(service);

verify(service,times(2)).serve(valueCaptor.capture());

// AssertionFailedError: expected:<[1, 2]> but was:<[2, 2]>"

assertEquals(Arrays.asList(new MutableInt(1), 

new MutableInt(2)),valueCaptor.getAllValues());

}

 

        /**

* Success with Answer

*/

public void testMutableArgsWithDoAnswer() {

Caller caller = new Caller();

doAnswer(new CaptureArgumentsWrittenAsMutableInt<Void>()).

when(service).serve(any(MutableInt.class));

caller.callService(service);

verify(service,times(2)).serve(any(MutableInt.class));

// Works!

assertEquals(new MutableInt(1),multiValuesWritten.get(0));

assertEquals(new MutableInt(2),multiValuesWritten.get(1));

}

/**

* Captures Arguments to the Service.serve() method:

* - Multiple calls to serve() happen from the same caller

* - Along with reuse of MutableInt argument objects by the caller

* - Argument value is copied to a new MutableInt object & that's captured

* @param <Void>

*/

public class CaptureArgumentsWrittenAsMutableInt<Void> implements Answer<Void>{

public Void answer(InvocationOnMock invocation) {

Object[] args = invocation.getArguments();

multiValuesWritten.add(new MutableInt(args[0].toString()));

return null ;

}

}

}

Java Versions & Features

Visual summary of Java Features added since Java 9. Feature clusters show the focus areas over the years. 

• Initially (ver 9+) focus was on adding some scripting type features & stabilizing the big ticket features added previously.
• GC & Performance was in focus through the next several versions.
• Patterns with Switch, InstanceOf, Type, etc have comein since ver 11+.
• From 14+ Foreign Memory, Vector API, Unix Socket, etc various performant direct host I/O/ parallel computing features have made it in.
• Some syntactic additions like Module import, When clause, etc are part of the more recent releases.
  

References:

• Diagram's datasheet
• https://medium.com/@chandantechie/comprehensive-list-of-java-versions-with-key-features-and-upcoming-releases-54be35646cca
• https://docs.oracle.com/en/java/javase/23/language/java-language-changes.html#GUID-6459681C-6881-45D8-B0DB-395D1BD6DB9B
• https://en.wikipedia.org/wiki/Java_version_history
  
• https://www.marcobehler.com/guides/a-guide-to-java-versions-and-features#_java_features_8_20
• https://www.javatpoint.com/java-versions
• https://howtodoinjava.com/series/java-versions-features/

Java Classloaders

Key things about Java classloaders:
1. There is a hierarchy among classloaders:
Bootstrap <---|
                      Extension <--|
                                          System <---|
Custom

- Child classloaders (typically) delegate class loading to parent classloaders. child class loader’s findClass() method is not called if the parent classloader can load the class.
- Custom classloaders can override the default delegation chain to a certain extent.
- Due to the delegation chain of classloaders, ensure classes to be loaded by custom classloaders are not present on the system class path, boot class path, or extension class path.

2. Bootstrap classloader is a special classloader included with the JVM, written in native code. Bootstrap classloader is tasked with loading all core classes part of the jre.
None of the other classloaders can override the Bootstrap classloader's behvaiour.

3. All other classloaders are written in Java.

4. Extension classloader loads classes from the extension directories: JAVA_HOME/jre/lib/ext/

5. A more popular alternative to using the Extension classloader, is to use the System classloader which loads classes from the CLASSPATH environment variable location.

6. Finally, Custom classloaders can be written to override certain defaults like delegating classloading to parents, etc. Custom classloaders is commonly used by Application servers (such as Tomcat).

7. Separate Namespaces per Classloader:
Same class loaded by two different classloaders, are considered different. Trying to cast an object of one class (loaded by classloader 1) to a reference of the other (loaded by classloader 2, though identical in terms of its fully qualified class name) will result in a ClassCastException.

8. Lazy loading and Caching of Classes:
Classloaders load classes lazily. Once loaded classloaders cache all previously loaded classes for the duration of the JVM.

Key Methods of Classloaders:
To be detailed..

Dynamic Reloading of Classes:
Due to the non-overridable behaviour of caching of classes by classloaders, reloading of classes within a running JVM poses problems. To reload a class dynamically (a common use case for app. servers), a new instance of the classloader itself needs to be created.
Once the earlier classsloader is orphaned/ garbage, classes loaded & cached by it (and reachable only via the now GC'd classloader) also become garbage, which can then be collected by the GC.

Java Generics PECS & Get-Put Principle, Non-Reifiable Types & Erasure, Covariance & Contravariance

Joshua Bloch's in the book Effective Java has introduced the PECS (Producer Extends Consumer Super) mnemonic for managing type hierarchies via Java Generics, i.e. Covariance & Contravariance for Generics.

Covariance & Contravariance

Covariance is a subtyping principle where a child/ sub-type can be used in place of parent/ super-type. In Java land this is seen with Arrays, Method Overriding (Java 5 onwards) & Generics Extends.

Contravariance is the reverse, where a parent/ super-type can be used in place of a sub-type. In Java, contravariance is seen with Generics Super.

PECS

Now coming back to PECS, also referred to as the Get-Put, principle of Generics. The key thing about PECS is that it is defined from the persepective of the Collection object in focus, and not the caller/ client using the Collection object.

Case 1: When the Collection is being used to retrieve existing (previously added) data, from the Collection's perspective it is a Producer of data. As per PECS it needs to Extend.
The caller in this case can be sure that results have objects who's parent is MyParent. So the caller can safely iterate over results as an immutable collection of MyParent objects.

However, any kind of addition into the Collection <? extends MyParent> is unsafe. MyParent can have any number of subtypes (MyChild1, MyChild2, etc.) and there is no way for the caller application to know this specific subtype. So at this stage any addition is not allowed.

Case 2: When the Collection is being used to store new data, from the Collection's perspective it is a Consumer of data. As per PECS it needs to define Super.

Type Erasure & Non-Reifiable Nature of Generics

Java Generics is implemented through erasure entirely at the compiler level. The type info. is discarded from within the bytecodes (i.e. List<String> & List have identical bytecodes). This also causes the non-reifiable behaviour of Generic types, i.e. the inability to determine the type info. from the bytecode at runtime. (Note that Arrays are reifiable as shown in E.g. 2 above).

The PECS rule is therefore to make the compiler operate defensively.  Once type info has been associated with a given Collection via Generics, the PECS restriction is there to enable compile time detection of potential unsafe usage.