Just find out workaround with this - create python logger with

logging.getLogger('my_logger')

, register it in

dagster.yaml

as

managed_python_loggers

and pass it to the function - will solve asset materialisation issue. https://docs.dagster.io/concepts/logging/python-logging

however it is interesting case and worth investigation, because of silent ignorance of assets de facto and tests passing can create hidden bugs that will be hard to debug

Hey Stepan, unfortunately I wasn't able to reproduce the issue you describe. I wrote up a simple test based around your example, feel free to run it. But from what I can see, 6 asset materialization events are logged by the op, which is what one would expect. Where exactly are you seeing the behavioral inconsistency? Something that could be throwing things off is that in this example, it's always writing to the same asset keys, so each might not be picked up as separate in dagit. If you switched asset keys for every asset materialization, or used partitions along with the asset materializatons, perhaps that might fix some confusion? Example test below:

import time
from concurrent.futures import ThreadPoolExecutor
from dagster import op, AssetMaterialization, AssetKey, Nothing, Out, job


def work(i: int, logger) -> int:
    time.sleep(3)
    return logger


def work_logger(i: int, logger) -> int:
    <http://logger.info|logger.info>(f'working {i}')
    time.sleep(3)
    return logger


def test_op():
    @op(out=Out(Nothing))
    def hello(context):
        """
        An op definition. This example op outputs a single string.
        For more hints about writing Dagster ops, see our documentation overview on Ops:
        <https://docs.dagster.io/concepts/ops-jobs-graphs/ops>
        """

        futures = []
        executor = ThreadPoolExecutor(1)
        for i in range(3):
            f = executor.submit(work, i, context.log)
            futures.append(f)

        for f in futures:
            <http://context.log.info|context.log.info>(f"waiting future done")
            res = f.result()
            yield AssetMaterialization(
                asset_key=AssetKey('asset'),
                metadata={"logger": str(res)},
            )
            <http://context.log.info|context.log.info>(f"future done {res}")

        context.log.warning("Starting jobs with logger in futures")

        futures = []
        for i in range(3):
            f = executor.submit(work_logger, i, context.log)
            futures.append(f)

        for f in futures:
            <http://context.log.info|context.log.info>(f"waiting future done")
            res = f.result()
            yield AssetMaterialization(
                asset_key=AssetKey('asset'),
                metadata={"logger": str(res)},
            )
            <http://context.log.info|context.log.info>(f"future done {res}")

        return Nothing

    @job
    def the_job():
        hello()

    result = the_job.execute_in_process()
    assert len(result.asset_materializations_for_node("hello")) == 6

Recently I have recreated asset_sensor behaviour. It will yield SkipReason on every asset and run nothing, but thats how we can count sensor cursor. (commit) After pipeline run - the sensor cursor is 4 and not 6 as expected

okay, was able to reproduce via dagit. Definitely a weird scenario that only seems to be brought about when logging inside the thread.