Unexpected error [ConcurrentModificationException]

[SalvadorGJ]

Bug report

My pipeline finishes abruptly with a non-detailed error when I try to resume it.

Expected behavior and actual behavior

I have the following channel definition

// Map reads to the genome
truncated_fastq_ch.fastq
        .combine(genome_index_ch)
        .map { sample_info, read1, read2, genome_info, index -> 
            sample_info.put('genomeName', "${genome_info['genomeName']}")
            return [sample_info, read1, read2, index]}
        .set { map_input_ch }

If I use .view() I get the expected behavior. Here is an example of the items in the channel map_input_ch

[[sampleName:CAUR4ANXX_2, sampleDataset:CAUR4ANXX, restrictionEnzime:^GATC,DpnII, genomeName:AmexG_v6], /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/6f/bd069037a1cfbe22ac571d5029df60/CAUR4ANXX_2_20170825B_20170830_left.part_003.fastq.trimmed_1P.trunc.fastq.gz, /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/6f/bd069037a1cfbe22ac571d5029df60/CAUR4ANXX_2_20170825B_20170830_left.part_003.fastq.trimmed_2P.trunc.fastq.gz, [/scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/e9/1689a4d49ae7342d8b7de59e27d493/genome_index.1.bt2l, /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/e9/1689a4d49ae7342d8b7de59e27d493/genome_index.2.bt2l, /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/e9/1689a4d49ae7342d8b7de59e27d493/genome_index.3.bt2l, /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/e9/1689a4d49ae7342d8b7de59e27d493/genome_index.4.bt2l, /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/e9/1689a4d49ae7342d8b7de59e27d493/genome_index.rev.1.bt2l, /scratch-cbe/users/salvador.gonzales/1_AmexGenomeAnnotation/0_AmexGenomeUpgrade/4_HiC_Processing/e9/1689a4d49ae7342d8b7de59e27d493/genome_index.rev.2.bt2l]]

Steps to reproduce the problem

Then when I run the following process the pipeline stops almost immediately,

process map_reads {
    executor = 'slurm' 
    cpus = 8
    time = { 1.hour * task.attempt }
    memory = { task.attempt < 2 ? 100.GB : 150.GB }
    errorStrategy = { 'retry' }
    maxRetries = 1
    maxForks = 1000
    clusterOptions = { '--qos=short --partition=c' }
    container = 'docker://docker.artifactory.imp.ac.at/tanakalab/hicup:master'

    input:
    tuple val(sample_info), path(read1), path(read2), path(index)

    output:
    tuple val(sample_info), path("*.pair.bam"), emit: bam
    tuple val(sample_info), path('hicup_mapper_summary*'), emit: summary

    script:
    """
    # Align read pairs independently to a reference genome
    hicup_mapper \\
        --bowtie2 /usr/bin/bowtie2 \\
        --index genome_index \\
        --threads ${task.cpus} \\
        --minlen 20 \\
        --mapq 30 \\
        --scorediff 10 \\
        --zip \\
        ${read1} ${read2}

    # Generate summary
    SUMMARY_FILE=\$(find . -name "hicup_mapper_summary*.txt")
    head -n 1 \${SUMMARY_FILE} > tmp
    grep ${read1} \${SUMMARY_FILE} >> tmp
    grep ${read2} \${SUMMARY_FILE} >> tmp
    rm \${SUMMARY_FILE}
    mv tmp \${SUMMARY_FILE}
    """
}

mapped_reads_ch = map_reads(map_input_ch)

Program output

ERROR ~ Unexpected error [ConcurrentModificationException]

issue_250128.nextflow.log

Environment

• Nextflow version: 23.10.1.5893
• Java version: [?]
• Operating system: Linux
• Bash version: GNU bash, version 4.2.46(2)-release (x86_64-redhat-linux-gnu)

Additional context

nextflow.enable.dsl = 2

[bentsherman]

It might be happening because you are modifying sample_info in the map operation. Operators run concurrently, so you generally shouldn't modify operator inputs (or process inputs for that matter) because it can cause a race condition (like the concurrent modification exception).

Try this instead:

        .map { sample_info, read1, read2, genome_info, index -> 
            [ sample_info + [genomeName: "${genome_info['genomeName']}"], read1, read2, index]
        }

[SalvadorGJ]

Hi @bentsherman ,

Thank you, the solution worked properly! Although I just made a small change to avoid the ERROR ~ No such variable: genomeName:

.map { sample_info, read1, read2, genome_info, index -> 
            [sample_info + ['genomeName': "${genome_info['genomeName']}"], read1, read2, index]}

[SalvadorGJ]

Hi,

I'm reopening this issue because it seems that the + operator doesn't preserve the groupKey in sample_info. This was previously reported in this issue. I was trying with this other way but I got the following error:

truncated_fastq_ch.fastq
        .combine(genome_index_ch)
        .map { sample_info, read1, read2, genome_info, index -> 
            def mapCopy = sample_info.target.clone()
            sample_info.put('genomeName', "${genome_info['genomeName']}")
            return [mapCopy, read1, read2, index]}
        .map { sample_info, read1, read2, index -> [ groupKey(sample_info, read1.size()), read1, read2, index ] }
        .transpose()
        .set { alignment_input_ch }

ERROR ~ Invalid method invocation `groupKey` with arguments: [sampleID:331733, series:2253CNLT1, sampleName:CSDCT, replicate:2, genomeName:231106_filt_100kb_contigs] (nextflow.extension.GroupKey), 1703197641 (java.lang.Long) on Nextflow type

Any suggestions?

[bentsherman]

In that case I suggest you defer the groupKey until right before the groupTuple operation. Just keep the group size alongside the sample_info or as a key-value within it. That way you don't have to unwrap/rewrap the groupKey every time you want to do a transformation