Fix MDC scope (#19501)

* Fix MDC scope * format

Author: benmoriceau

airbyte-commons-worker/src/main/java/io/airbyte/workers/internal/DefaultAirbyteDestination.java

@@ -36,7 +36,7 @@
 public class DefaultAirbyteDestination implements AirbyteDestination {
 
  private static final Logger LOGGER = LoggerFactory.getLogger(DefaultAirbyteDestination.class);
- private static final MdcScope.Builder CONTAINER_LOG_MDC_BUILDER = new Builder()
+ public static final MdcScope.Builder CONTAINER_LOG_MDC_BUILDER = new Builder()
  .setLogPrefix("destination")
  .setPrefixColor(Color.YELLOW_BACKGROUND);
 

airbyte-commons-worker/src/main/java/io/airbyte/workers/internal/DefaultAirbyteSource.java

@@ -39,7 +39,7 @@ public class DefaultAirbyteSource implements AirbyteSource {
  private static final Duration HEARTBEAT_FRESH_DURATION = Duration.of(5, ChronoUnit.MINUTES);
  private static final Duration GRACEFUL_SHUTDOWN_DURATION = Duration.of(1, ChronoUnit.MINUTES);
 
- private static final MdcScope.Builder CONTAINER_LOG_MDC_BUILDER = new Builder()
+ public static final MdcScope.Builder CONTAINER_LOG_MDC_BUILDER = new Builder()
  .setLogPrefix("source")
  .setPrefixColor(Color.BLUE_BACKGROUND);
 

airbyte-container-orchestrator/src/main/java/io/airbyte/container_orchestrator/ReplicationJobOrchestrator.java

@@ -12,6 +12,7 @@
 import datadog.trace.api.Trace;
 import io.airbyte.commons.features.FeatureFlags;
 import io.airbyte.commons.json.Jsons;
+import io.airbyte.commons.logging.MdcScope;
 import io.airbyte.commons.protocol.AirbyteMessageSerDeProvider;
 import io.airbyte.commons.protocol.AirbyteMessageVersionedMigratorFactory;
 import io.airbyte.commons.temporal.TemporalUtils;
@@ -120,7 +121,8 @@ public Optional<String> runJob() throws Exception {
  final AirbyteSource airbyteSource =
  WorkerConstants.RESET_JOB_SOURCE_DOCKER_IMAGE_STUB.equals(sourceLauncherConfig.getDockerImage()) ? new EmptyAirbyteSource(
  featureFlags.useStreamCapableState())
- : new DefaultAirbyteSource(sourceLauncher, getStreamFactory(sourceLauncherConfig.getProtocolVersion()));
+ : new DefaultAirbyteSource(sourceLauncher,
+ getStreamFactory(sourceLauncherConfig.getProtocolVersion(), DefaultAirbyteSource.CONTAINER_LOG_MDC_BUILDER));
 
  MetricClientFactory.initialize(MetricEmittingApps.WORKER);
  final MetricClient metricClient = MetricClientFactory.getMetricClient();
@@ -132,7 +134,8 @@ public Optional<String> runJob() throws Exception {
  Math.toIntExact(jobRunConfig.getAttemptId()),
  airbyteSource,
  new NamespacingMapper(syncInput.getNamespaceDefinition(), syncInput.getNamespaceFormat(), syncInput.getPrefix()),
- new DefaultAirbyteDestination(destinationLauncher, getStreamFactory(destinationLauncherConfig.getProtocolVersion()),
+ new DefaultAirbyteDestination(destinationLauncher, getStreamFactory(destinationLauncherConfig.getProtocolVersion(),
+ DefaultAirbyteDestination.CONTAINER_LOG_MDC_BUILDER),
  new VersionedAirbyteMessageBufferedWriterFactory(serDeProvider, migratorFactory, destinationLauncherConfig.getProtocolVersion())),
  new AirbyteMessageTracker(),
  new RecordSchemaValidator(WorkerUtils.mapStreamNamesToSchemas(syncInput)),
@@ -146,10 +149,10 @@ public Optional<String> runJob() throws Exception {
  return Optional.of(Jsons.serialize(replicationOutput));
  }
 
- private AirbyteStreamFactory getStreamFactory(final Version protocolVersion) {
+ private AirbyteStreamFactory getStreamFactory(final Version protocolVersion, final MdcScope.Builder mdcScope) {
  return protocolVersion != null
- ? new VersionedAirbyteStreamFactory<>(serDeProvider, migratorFactory, protocolVersion)
- : new DefaultAirbyteStreamFactory();
+ ? new VersionedAirbyteStreamFactory(serDeProvider, migratorFactory, protocolVersion, mdcScope)
+ : new DefaultAirbyteStreamFactory(mdcScope);
  }
 
 }

airbyte-workers/src/main/java/io/airbyte/workers/temporal/sync/ReplicationActivityImpl.java

@@ -276,7 +276,8 @@ private CheckedSupplier<Worker<StandardSyncInput, ReplicationOutput>, Exception>
  WorkerConstants.RESET_JOB_SOURCE_DOCKER_IMAGE_STUB.equals(sourceLauncherConfig.getDockerImage())
  ? new EmptyAirbyteSource(featureFlags.useStreamCapableState())
  : new DefaultAirbyteSource(sourceLauncher,
- new VersionedAirbyteStreamFactory<>(serDeProvider, migratorFactory, sourceLauncherConfig.getProtocolVersion()));
+ new VersionedAirbyteStreamFactory<>(serDeProvider, migratorFactory, sourceLauncherConfig.getProtocolVersion(),
+ DefaultAirbyteSource.CONTAINER_LOG_MDC_BUILDER));
  MetricClientFactory.initialize(MetricEmittingApps.WORKER);
  final MetricClient metricClient = MetricClientFactory.getMetricClient();
  final WorkerMetricReporter metricReporter = new WorkerMetricReporter(metricClient, sourceLauncherConfig.getDockerImage());
@@ -287,7 +288,8 @@ private CheckedSupplier<Worker<StandardSyncInput, ReplicationOutput>, Exception>
  airbyteSource,
  new NamespacingMapper(syncInput.getNamespaceDefinition(), syncInput.getNamespaceFormat(), syncInput.getPrefix()),
  new DefaultAirbyteDestination(destinationLauncher,
- new VersionedAirbyteStreamFactory<>(serDeProvider, migratorFactory, destinationLauncherConfig.getProtocolVersion()),
+ new VersionedAirbyteStreamFactory<>(serDeProvider, migratorFactory, destinationLauncherConfig.getProtocolVersion(),
+ DefaultAirbyteDestination.CONTAINER_LOG_MDC_BUILDER),
  new VersionedAirbyteMessageBufferedWriterFactory(serDeProvider, migratorFactory, destinationLauncherConfig.getProtocolVersion())),
  new AirbyteMessageTracker(),
  new RecordSchemaValidator(WorkerUtils.mapStreamNamesToSchemas(syncInput)),