Get exchangeId into the exchange sink/source profile output

craigtaverner · craigtaverner · commit 7fb4075db233 · 2025-02-27T18:06:00.000+01:00
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeService.java
@@ -123,7 +123,12 @@ public void registerTransportHandler(TransportService transportService) {
      * @throws IllegalStateException if a sink handler for the given id already exists
      */
     public ExchangeSinkHandler createSinkHandler(String exchangeId, int maxBufferSize) {
-        ExchangeSinkHandler sinkHandler = new ExchangeSinkHandler(blockFactory, maxBufferSize, threadPool.relativeTimeInMillisSupplier());
+        ExchangeSinkHandler sinkHandler = new ExchangeSinkHandler(
+            exchangeId,
+            blockFactory,
+            maxBufferSize,
+            threadPool.relativeTimeInMillisSupplier()
+        );
         if (sinks.putIfAbsent(exchangeId, sinkHandler) != null) {
             throw new IllegalStateException("sink exchanger for id [" + exchangeId + "] already exists");
         }
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSink.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSink.java
@@ -40,4 +40,9 @@ public interface ExchangeSink {
      * Whether the sink is blocked on adding more pages
      */
     IsBlockedResult waitForWriting();
+
+    /**
+     * Session ID for this exchange
+     */
+    String exchangeId();
 }
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSinkHandler.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSinkHandler.java
@@ -40,9 +40,11 @@ public final class ExchangeSinkHandler {
     private final SubscribableListener<Void> completionFuture;
     private final LongSupplier nowInMillis;
     private final AtomicLong lastUpdatedInMillis;
+    private final String exchangeId;
     private final BlockFactory blockFactory;
 
-    public ExchangeSinkHandler(BlockFactory blockFactory, int maxBufferSize, LongSupplier nowInMillis) {
+    public ExchangeSinkHandler(String exchangeId, BlockFactory blockFactory, int maxBufferSize, LongSupplier nowInMillis) {
+        this.exchangeId = exchangeId;
         this.blockFactory = blockFactory;
         this.buffer = new ExchangeBuffer(maxBufferSize);
         this.completionFuture = SubscribableListener.newForked(buffer::addCompletionListener);
@@ -51,11 +53,13 @@ public ExchangeSinkHandler(BlockFactory blockFactory, int maxBufferSize, LongSup
     }
 
     private class ExchangeSinkImpl implements ExchangeSink {
+        private final String exchangeId;
         boolean finished;
         private final Runnable onPageFetched;
         private final SubscribableListener<Void> onFinished = new SubscribableListener<>();
 
-        ExchangeSinkImpl(Runnable onPageFetched) {
+        ExchangeSinkImpl(String exchangeId, Runnable onPageFetched) {
+            this.exchangeId = exchangeId;
             this.onPageFetched = onPageFetched;
             onChanged();
             buffer.addCompletionListener(onFinished);
@@ -96,6 +100,11 @@ public void addCompletionListener(ActionListener<Void> listener) {
         public IsBlockedResult waitForWriting() {
             return buffer.waitForWriting();
         }
+
+        @Override
+        public String exchangeId() {
+            return exchangeId;
+        }
     }
 
     /**
@@ -168,7 +177,7 @@ private void notifyListeners() {
      * @see ExchangeSinkOperator
      */
     public ExchangeSink createExchangeSink(Runnable onPageFetched) {
-        return new ExchangeSinkImpl(onPageFetched);
+        return new ExchangeSinkImpl(exchangeId, onPageFetched);
     }
 
     /**
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSinkOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSinkOperator.java
@@ -94,7 +94,7 @@ public void close() {
 
     @Override
     public String toString() {
-        return "ExchangeSinkOperator";
+        return "ExchangeSinkOperator[" + sink.exchangeId() + "]";
     }
 
     @Override
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSource.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSource.java
@@ -39,4 +39,9 @@ public interface ExchangeSource {
      * Allows callers to stop reading from the source when it's blocked
      */
     IsBlockedResult waitForReading();
+
+    /**
+     * Session ID for this exchange
+     */
+    String exchangeId();
 }
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSourceHandler.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSourceHandler.java
@@ -30,6 +30,7 @@
  * @see #addRemoteSink(RemoteSink, boolean, Runnable, int, ActionListener)
  */
 public final class ExchangeSourceHandler {
+    private final String exchangeId;
     private final ExchangeBuffer buffer;
     private final Executor fetchExecutor;
 
@@ -49,7 +50,8 @@ public final class ExchangeSourceHandler {
      *                           which could otherwise be allocated for other purposes.
      * @param fetchExecutor      the executor used to fetch pages.
      */
-    public ExchangeSourceHandler(int maxBufferSize, Executor fetchExecutor) {
+    public ExchangeSourceHandler(String exchangeId, int maxBufferSize, Executor fetchExecutor) {
+        this.exchangeId = exchangeId;
         this.buffer = new ExchangeBuffer(maxBufferSize);
         this.fetchExecutor = fetchExecutor;
         this.outstandingSinks = new PendingInstances(() -> buffer.finish(false));
@@ -63,9 +65,11 @@ private void checkFailure() {
     }
 
     private class ExchangeSourceImpl implements ExchangeSource {
+        private final String exchangeId1;
         private boolean finished;
 
-        ExchangeSourceImpl() {
+        ExchangeSourceImpl(String exchangeId) {
+            exchangeId1 = exchangeId;
             outstandingSources.trackNewInstance();
         }
 
@@ -86,6 +90,11 @@ public IsBlockedResult waitForReading() {
             return buffer.waitForReading();
         }
 
+        @Override
+        public String exchangeId() {
+            return exchangeId1;
+        }
+
         @Override
         public void finish() {
             if (finished == false) {
@@ -106,7 +115,7 @@ public int bufferSize() {
      * @see ExchangeSinkOperator
      */
     public ExchangeSource createExchangeSource() {
-        return new ExchangeSourceImpl();
+        return new ExchangeSourceImpl(exchangeId);
     }
 
     /**
diff --git a/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSourceOperator.java b/x-pack/plugin/esql/compute/src/main/java/org/elasticsearch/compute/operator/exchange/ExchangeSourceOperator.java
@@ -89,7 +89,7 @@ public void close() {
 
     @Override
     public String toString() {
-        return "ExchangeSourceOperator";
+        return "ExchangeSourceOperator[" + source.exchangeId() + "]";
     }
 
     @Override
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/DriverTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/DriverTests.java
@@ -289,7 +289,7 @@ public void testEarlyTermination() {
             final var sourceOperator = new CannedSourceOperator(inPages.iterator());
             final int maxAllowedRows = between(1, 100);
             final AtomicInteger processedRows = new AtomicInteger(0);
-            var sinkHandler = new ExchangeSinkHandler(driverContext.blockFactory(), positions, System::currentTimeMillis);
+            var sinkHandler = new ExchangeSinkHandler("test", driverContext.blockFactory(), positions, System::currentTimeMillis);
             var sinkOperator = new ExchangeSinkOperator(sinkHandler.createExchangeSink(() -> {}), Function.identity());
             final var delayOperator = new EvalOperator(driverContext.blockFactory(), new EvalOperator.ExpressionEvaluator() {
                 @Override
@@ -324,8 +324,8 @@ public void testResumeOnEarlyFinish() throws Exception {
         DriverContext driverContext = driverContext();
         ThreadPool threadPool = threadPool();
         try {
-            var sourceHandler = new ExchangeSourceHandler(between(1, 5), threadPool.executor("esql"));
-            var sinkHandler = new ExchangeSinkHandler(driverContext.blockFactory(), between(1, 5), System::currentTimeMillis);
+            var sourceHandler = new ExchangeSourceHandler("test", between(1, 5), threadPool.executor("esql"));
+            var sinkHandler = new ExchangeSinkHandler("test", driverContext.blockFactory(), between(1, 5), System::currentTimeMillis);
             var sourceOperator = new ExchangeSourceOperator(sourceHandler.createExchangeSource());
             var sinkOperator = new ExchangeSinkOperator(sinkHandler.createExchangeSink(() -> {}), Function.identity());
             Driver driver = TestDriverFactory.create(driverContext, sourceOperator, List.of(), sinkOperator);
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/ForkingOperatorTestCase.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/ForkingOperatorTestCase.java
@@ -205,11 +205,16 @@ List<Driver> createDriversForInput(List<Page> input, List<Page> results, boolean
         Collection<List<Page>> splitInput = randomSplits(input, randomIntBetween(2, 4));
         BlockFactory factory = blockFactory();
         ExchangeSinkHandler sinkExchanger = new ExchangeSinkHandler(
+            "test",
             factory,
             randomIntBetween(2, 10),
             threadPool.relativeTimeInMillisSupplier()
         );
-        ExchangeSourceHandler sourceExchanger = new ExchangeSourceHandler(randomIntBetween(1, 4), threadPool.executor(ESQL_TEST_EXECUTOR));
+        ExchangeSourceHandler sourceExchanger = new ExchangeSourceHandler(
+            "test",
+            randomIntBetween(1, 4),
+            threadPool.executor(ESQL_TEST_EXECUTOR)
+        );
         sourceExchanger.addRemoteSink(
             sinkExchanger::fetchPageAsync,
             randomBoolean(),
diff --git a/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/exchange/ExchangeServiceTests.java b/x-pack/plugin/esql/compute/src/test/java/org/elasticsearch/compute/operator/exchange/ExchangeServiceTests.java
@@ -102,11 +102,11 @@ public void testBasic() throws Exception {
         for (int i = 0; i < pages.length; i++) {
             pages[i] = new Page(blockFactory.newConstantIntBlockWith(i, 2));
         }
-        ExchangeSinkHandler sinkExchanger = new ExchangeSinkHandler(blockFactory, 2, threadPool.relativeTimeInMillisSupplier());
+        ExchangeSinkHandler sinkExchanger = new ExchangeSinkHandler("test", blockFactory, 2, threadPool.relativeTimeInMillisSupplier());
         AtomicInteger pagesAddedToSink = new AtomicInteger();
         ExchangeSink sink1 = sinkExchanger.createExchangeSink(pagesAddedToSink::incrementAndGet);
         ExchangeSink sink2 = sinkExchanger.createExchangeSink(pagesAddedToSink::incrementAndGet);
-        ExchangeSourceHandler sourceExchanger = new ExchangeSourceHandler(3, threadPool.executor(ESQL_TEST_EXECUTOR));
+        ExchangeSourceHandler sourceExchanger = new ExchangeSourceHandler("test", 3, threadPool.executor(ESQL_TEST_EXECUTOR));
         ExchangeSource source = sourceExchanger.createExchangeSource();
         AtomicInteger pagesAddedToSource = new AtomicInteger();
         PlainActionFuture<Void> remoteSinkFuture = new PlainActionFuture<>();
@@ -358,7 +358,7 @@ private static Driver createDriver(
 
     public void testConcurrentWithHandlers() {
         BlockFactory blockFactory = blockFactory();
-        var sourceExchanger = new ExchangeSourceHandler(randomExchangeBuffer(), threadPool.executor(ESQL_TEST_EXECUTOR));
+        var sourceExchanger = new ExchangeSourceHandler("test", randomExchangeBuffer(), threadPool.executor(ESQL_TEST_EXECUTOR));
         PlainActionFuture<Void> remoteSinksFuture = new PlainActionFuture<>();
         try (RefCountingListener refs = new RefCountingListener(remoteSinksFuture)) {
             List<ExchangeSinkHandler> sinkHandlers = new ArrayList<>();
@@ -367,7 +367,12 @@ public void testConcurrentWithHandlers() {
                 if (sinkHandlers.isEmpty() == false && randomBoolean()) {
                     sinkHandler = randomFrom(sinkHandlers);
                 } else {
-                    sinkHandler = new ExchangeSinkHandler(blockFactory, randomExchangeBuffer(), threadPool.relativeTimeInMillisSupplier());
+                    sinkHandler = new ExchangeSinkHandler(
+                        "test",
+                        blockFactory,
+                        randomExchangeBuffer(),
+                        threadPool.relativeTimeInMillisSupplier()
+                    );
                     sourceExchanger.addRemoteSink(
                         sinkHandler::fetchPageAsync,
                         randomBoolean(),
@@ -396,7 +401,7 @@ public void testConcurrentWithHandlers() {
 
     public void testExchangeSourceContinueOnFailure() {
         BlockFactory blockFactory = blockFactory();
-        var exchangeSourceHandler = new ExchangeSourceHandler(randomExchangeBuffer(), threadPool.executor(ESQL_TEST_EXECUTOR));
+        var exchangeSourceHandler = new ExchangeSourceHandler("test", randomExchangeBuffer(), threadPool.executor(ESQL_TEST_EXECUTOR));
         final int maxInputSeqNo = rarely() ? -1 : randomIntBetween(0, 50_000);
         final int maxOutputSeqNo = rarely() ? -1 : randomIntBetween(0, 50_000);
         Set<Integer> expectedSeqNos = ConcurrentCollections.newConcurrentSet();
@@ -407,7 +412,12 @@ public void testExchangeSourceContinueOnFailure() {
         PlainActionFuture<Void> remoteSinksFuture = new PlainActionFuture<>();
         try (RefCountingListener refs = new RefCountingListener(remoteSinksFuture)) {
             Supplier<ExchangeSink> exchangeSink = () -> {
-                var sinkHandler = new ExchangeSinkHandler(blockFactory, randomExchangeBuffer(), threadPool.relativeTimeInMillisSupplier());
+                var sinkHandler = new ExchangeSinkHandler(
+                    "test",
+                    blockFactory,
+                    randomExchangeBuffer(),
+                    threadPool.relativeTimeInMillisSupplier()
+                );
                 int failAfter = randomBoolean() ? Integer.MAX_VALUE : randomIntBetween(0, 100);
                 AtomicInteger fetched = new AtomicInteger();
                 int instance = randomIntBetween(1, 3);
@@ -472,7 +482,7 @@ public void testClosingSinks() {
         IntBlock block2 = blockFactory.newConstantIntBlockWith(1, 2);
         Page p1 = new Page(block1);
         Page p2 = new Page(block2);
-        ExchangeSinkHandler sinkExchanger = new ExchangeSinkHandler(blockFactory, 2, threadPool.relativeTimeInMillisSupplier());
+        ExchangeSinkHandler sinkExchanger = new ExchangeSinkHandler("test", blockFactory, 2, threadPool.relativeTimeInMillisSupplier());
         ExchangeSink sink = sinkExchanger.createExchangeSink(() -> {});
         sink.addPage(p1);
         sink.addPage(p2);
@@ -487,7 +497,7 @@ public void testClosingSinks() {
     }
 
     public void testFinishEarly() throws Exception {
-        ExchangeSourceHandler sourceHandler = new ExchangeSourceHandler(20, threadPool.generic());
+        ExchangeSourceHandler sourceHandler = new ExchangeSourceHandler("test", 20, threadPool.generic());
         Semaphore permits = new Semaphore(between(1, 5));
         BlockFactory blockFactory = blockFactory();
         Queue<Page> pages = ConcurrentCollections.newQueue();
@@ -558,7 +568,7 @@ public void testConcurrentWithTransportActions() {
         try (exchange0; exchange1; node0; node1) {
             String exchangeId = "exchange";
             Task task = new Task(1, "", "", "", null, Collections.emptyMap());
-            var sourceHandler = new ExchangeSourceHandler(randomExchangeBuffer(), threadPool.executor(ESQL_TEST_EXECUTOR));
+            var sourceHandler = new ExchangeSourceHandler(exchangeId, randomExchangeBuffer(), threadPool.executor(ESQL_TEST_EXECUTOR));
             ExchangeSinkHandler sinkHandler = exchange1.createSinkHandler(exchangeId, randomExchangeBuffer());
             Transport.Connection connection = node0.getConnection(node1.getLocalNode());
             sourceHandler.addRemoteSink(
@@ -628,7 +638,7 @@ public void sendResponse(TransportResponse transportResponse) {
         try (exchange0; exchange1; node0; node1) {
             String exchangeId = "exchange";
             Task task = new Task(1, "", "", "", null, Collections.emptyMap());
-            var sourceHandler = new ExchangeSourceHandler(randomIntBetween(1, 128), threadPool.executor(ESQL_TEST_EXECUTOR));
+            var sourceHandler = new ExchangeSourceHandler(exchangeId, randomIntBetween(1, 128), threadPool.executor(ESQL_TEST_EXECUTOR));
             ExchangeSinkHandler sinkHandler = exchange1.createSinkHandler(exchangeId, randomIntBetween(1, 128));
             Transport.Connection connection = node0.getConnection(node1.getLocalNode());
             PlainActionFuture<Void> remoteSinkFuture = new PlainActionFuture<>();
@@ -662,7 +672,7 @@ public void sendResponse(TransportResponse transportResponse) {
     public void testNoCyclicException() throws Exception {
         PlainActionFuture<Void> future = new PlainActionFuture<>();
         try (EsqlRefCountingListener refs = new EsqlRefCountingListener(future)) {
-            var exchangeSourceHandler = new ExchangeSourceHandler(between(10, 100), threadPool.generic());
+            var exchangeSourceHandler = new ExchangeSourceHandler("test", between(10, 100), threadPool.generic());
             int numSinks = between(5, 10);
             for (int i = 0; i < numSinks; i++) {
                 RemoteSink remoteSink = (allSourcesFinished, listener) -> threadPool.schedule(
diff --git a/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionTaskIT.java b/x-pack/plugin/esql/src/internalClusterTest/java/org/elasticsearch/xpack/esql/action/EsqlActionTaskIT.java
@@ -104,6 +104,7 @@ public void testTaskContents() throws Exception {
             for (TaskInfo task : foundTasks) {
                 DriverStatus status = (DriverStatus) task.status();
                 assertThat(status.sessionId(), not(emptyOrNullString()));
+                String exchangeId = status.sessionId().replace("[n]", "");
                 String taskDescription = status.taskDescription();
                 for (OperatorStatus o : status.activeOperators()) {
                     logger.info("status {}", o);
@@ -138,15 +139,15 @@ public void testTaskContents() throws Exception {
                         valuesSourceReaders++;
                         continue;
                     }
-                    if (o.operator().equals("ExchangeSourceOperator")) {
+                    if (o.operator().startsWith("ExchangeSourceOperator[" + exchangeId + "]")) {
                         assertThat(taskDescription, either(equalTo("node_reduce")).or(equalTo("final")));
                         ExchangeSourceOperator.Status oStatus = (ExchangeSourceOperator.Status) o.status();
                         assertThat(oStatus.pagesWaiting(), greaterThanOrEqualTo(0));
                         assertThat(oStatus.pagesEmitted(), greaterThanOrEqualTo(0));
                         exchangeSources++;
                         continue;
                     }
-                    if (o.operator().equals("ExchangeSinkOperator")) {
+                    if (o.operator().startsWith("ExchangeSinkOperator[" + exchangeId + "]")) {
                         assertThat(taskDescription, either(equalTo("data")).or(equalTo("node_reduce")));
                         ExchangeSinkOperator.Status oStatus = (ExchangeSinkOperator.Status) o.status();
                         assertThat(oStatus.pagesReceived(), greaterThanOrEqualTo(0));
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ClusterComputeHandler.java
@@ -249,6 +249,7 @@ void runComputeOnRemoteCluster(
             return new ComputeResponse(profiles, took, r.totalShards, r.successfulShards, r.skippedShards, r.failedShards);
         }))) {
             var exchangeSource = new ExchangeSourceHandler(
+                globalSessionId,
                 configuration.pragmas().exchangeBufferSize(),
                 transportService.getThreadPool().executor(ThreadPool.Names.SEARCH)
             );
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/ComputeService.java
@@ -196,6 +196,7 @@ public void execute(
          */
         List<Attribute> outputAttributes = physicalPlan.output();
         var exchangeSource = new ExchangeSourceHandler(
+            sessionId,
             queryPragmas.exchangeBufferSize(),
             transportService.getThreadPool().executor(ThreadPool.Names.SEARCH)
         );
diff --git a/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeComputeHandler.java b/x-pack/plugin/esql/src/main/java/org/elasticsearch/xpack/esql/plugin/DataNodeComputeHandler.java
@@ -407,7 +407,7 @@ private void runComputeOnDataNode(
                 task.addListener(
                     () -> exchangeService.finishSinkHandler(externalId, new TaskCancelledException(task.getReasonCancelled()))
                 );
-                var exchangeSource = new ExchangeSourceHandler(1, esqlExecutor);
+                var exchangeSource = new ExchangeSourceHandler(externalId, 1, esqlExecutor);
                 exchangeSource.addRemoteSink(internalSink::fetchPageAsync, true, () -> {}, 1, ActionListener.noop());
                 var reductionListener = computeListener.acquireCompute();
                 computeService.runCompute(
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/CsvTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/CsvTests.java
@@ -633,8 +633,8 @@ void executeSubPlan(
             bigArrays,
             ByteSizeValue.ofBytes(randomLongBetween(1, BlockFactory.DEFAULT_MAX_BLOCK_PRIMITIVE_ARRAY_SIZE.getBytes() * 2))
         );
-        ExchangeSourceHandler exchangeSource = new ExchangeSourceHandler(between(1, 64), executor);
-        ExchangeSinkHandler exchangeSink = new ExchangeSinkHandler(blockFactory, between(1, 64), threadPool::relativeTimeInMillis);
+        ExchangeSourceHandler exchangeSource = new ExchangeSourceHandler("test", between(1, 64), executor);
+        ExchangeSinkHandler exchangeSink = new ExchangeSinkHandler("test", blockFactory, between(1, 64), threadPool::relativeTimeInMillis);
 
         LocalExecutionPlanner executionPlanner = new LocalExecutionPlanner(
             getTestName(),
diff --git a/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java b/x-pack/plugin/esql/src/test/java/org/elasticsearch/xpack/esql/optimizer/PhysicalPlanOptimizerTests.java

Original file line number	Diff line number	Diff line change
`@@ -94,7 +94,7 @@ public void close() {`
`94`	`94`
`95`	`95`	`@Override`
`96`	`96`	`public String toString() {`
`97`		`- return "ExchangeSinkOperator";`
	`97`	`+ return "ExchangeSinkOperator[" + sink.exchangeId() + "]";`
`98`	`98`	`}`
`99`	`99`
`100`	`100`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -89,7 +89,7 @@ public void close() {`
`89`	`89`
`90`	`90`	`@Override`
`91`	`91`	`public String toString() {`
`92`		`- return "ExchangeSourceOperator";`
	`92`	`+ return "ExchangeSourceOperator[" + source.exchangeId() + "]";`
`93`	`93`	`}`
`94`	`94`
`95`	`95`	`@Override`