{"record":{"id":"799e0a3ff9a50471","repo":"apache/hadoop","slug":"error-sending-data","errorCode":null,"errorMessage":"Error sending data","messagePattern":"Error sending data","errorType":"exception","errorClass":"MetricsException","httpStatus":null,"severity":"error","filePath":"hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java","lineNumber":180,"sourceCode":"\n    // Send the data to the Kafka broker. Here is an example of this data:\n    // {\"hostname\": \"...\", \"timestamp\": 1436913651516,\n    // \"date\": \"2015-6-14\",\"time\": \"22:40:51\",\"context\": \"yarn\",\"name\":\n    // \"QueueMetrics, \"running_0\": \"1\", \"running_60\": \"0\", \"running_300\": \"0\",\n    // \"running_1440\": \"0\", \"AppsSubmitted\": \"1\", \"AppsRunning\": \"1\",\n    // \"AppsPending\": \"0\", \"AppsCompleted\": \"0\", \"AppsKilled\": \"0\",\n    // \"AppsFailed\": \"0\", \"AllocatedMB\": \"134656\", \"AllocatedVCores\": \"132\",\n    // \"AllocatedContainers\": \"132\", \"AggregateContainersAllocated\": \"132\",\n    // \"AggregateContainersReleased\": \"0\", \"AvailableMB\": \"0\",\n    // \"AvailableVCores\": \"0\", \"PendingMB\": \"275456\", \"PendingVCores\": \"269\",\n    // \"PendingContainers\": \"269\", \"ReservedMB\": \"0\", \"ReservedVCores\": \"0\",\n    // \"ReservedContainers\": \"0\", \"ActiveUsers\": \"1\", \"ActiveApplications\": \"1\"}\n    Future<RecordMetadata> future = producer.send(data);\n    jsonLines.setLength(0);\n    try {\n      future.get();\n    } catch (InterruptedException e) {\n      throw new MetricsException(\"Error sending data\", e);\n    } catch (ExecutionException e) {\n      throw new MetricsException(\"Error sending data\", e);\n    }\n  }\n\n  @Override\n  public void flush() {\n    LOG.debug(\"Kafka seems not to have any flush() mechanism!\");\n  }\n\n  @Override\n  public void close() throws IOException {\n    // Close the producer and set it to null.\n    try {\n      producer.close();\n    } catch (RuntimeException e) {\n      throw new MetricsException(\"Error closing producer\", e);\n    } finally {","sourceCodeStart":162,"sourceCodeEnd":198,"githubUrl":"https://github.com/apache/hadoop/blob/2add9630210752f88ceb1bb74eb65e37bf41da8e/hadoop-tools/hadoop-kafka/src/main/java/org/apache/hadoop/metrics2/sink/KafkaSink.java#L162-L198","documentation":"In KafkaSink.putMetrics(), after producer.send(data) the sink blocks on future.get(). If the waiting thread is interrupted, InterruptedException is caught and rethrown as MetricsException('Error sending data'). Unlike the ExecutionException path (5098), this variant means the publishing thread was cancelled while waiting for Kafka's ack — not that Kafka rejected the record.","triggerScenarios":"The metrics publishing thread is interrupted while blocked in future.get(): daemon shutdown (metrics system stopping sink threads), long Kafka ack latency combined with thread pool shutdown/now, or explicit cancellation of the metrics timer thread.","commonSituations":"Rolling restarts or shutdown of NameNode/ResourceManager while metrics are in flight; Kafka slow or unreachable (request.required.acks=0 still involves metadata I/O) so get() blocks until the interrupt arrives; executor shutdownNow() during reconfiguration of metrics.","solutions":["Treat as shutdown noise if it appears only during daemon stop/restart — verify Kafka health separately","Fix underlying slowness: check broker availability and network from Hadoop nodes (acks latency), since a fast get() rarely overlaps interrupts","If it recurs at runtime (not shutdown), capture the stack trace to identify which component is interrupting the metrics thread"],"exampleFix":null,"handlingStrategy":"try-catch","validationCode":null,"typeGuard":null,"tryCatchPattern":"try {\n  sink.putMetrics(record);\n} catch (MetricsException e) {\n  if (e.getCause() instanceof InterruptedException) {\n    Thread.currentThread().interrupt(); // restore flag\n    // metrics thread cancelled (shutdown); stop publishing\n    return;\n  }\n  throw e;\n}","preventionTips":["Keep Kafka ack latency low so future.get() rarely overlaps shutdown interrupts","Restore the interrupt flag when catching this variant — the sink does not"],"tags":["hadoop","metrics2","kafka","interrupt","send"],"backgroundTag":"thread-interrupted","analyzedSha":"2add9630210752f88ceb1bb74eb65e37bf41da8e","analyzedAt":"2026-08-22T19:55:07.957Z","schemaVersion":2},"datasetVersion":"2026-08-22T20:17:22.307Z"}