[client] Improve: add detailed context to exception messages (#1806)

Author: CodeDrinks270

fluss-client/src/main/java/org/apache/fluss/client/table/scanner/TableScan.java

@@ -74,7 +74,11 @@ public Scan project(List<String> projectedColumnNames) {
             int index = rowType.getFieldIndex(projectedColumnNames.get(i));
             if (index < 0) {
                 throw new IllegalArgumentException(
-                        "Field " + projectedColumnNames.get(i) + " not found in table schema.");
+                        String.format(
+                                "Field '%s' not found in table schema. Available fields: %s, Table: %s",
+                                projectedColumnNames.get(i),
+                                rowType.getFieldNames(),
+                                tableInfo.getTablePath()));
             }
             columnIndexes[i] = index;
         }
@@ -89,7 +93,10 @@ public Scan limit(int rowNumber) {
     @Override
     public LogScanner createLogScanner() {
         if (limit != null) {
-            throw new UnsupportedOperationException("LogScanner doesn't support limit pushdown.");
+            throw new UnsupportedOperationException(
+                    String.format(
+                            "LogScanner doesn't support limit pushdown. Table: %s, requested limit: %d",
+                            tableInfo.getTablePath(), limit));
         }
         return new LogScannerImpl(
                 conn.getConfiguration(),
@@ -104,7 +111,9 @@ public LogScanner createLogScanner() {
     public BatchScanner createBatchScanner(TableBucket tableBucket) {
         if (limit == null) {
             throw new UnsupportedOperationException(
-                    "Currently, BatchScanner is only available when limit is set.");
+                    String.format(
+                            "Currently, BatchScanner is only available when limit is set. Table: %s, bucket: %s",
+                            tableInfo.getTablePath(), tableBucket));
         }
         return new LimitBatchScanner(
                 tableInfo, tableBucket, conn.getMetadataUpdater(), projectedColumns, limit);
@@ -114,7 +123,9 @@ public BatchScanner createBatchScanner(TableBucket tableBucket) {
     public BatchScanner createBatchScanner(TableBucket tableBucket, long snapshotId) {
         if (limit != null) {
             throw new UnsupportedOperationException(
-                    "Currently, SnapshotBatchScanner doesn't support limit pushdown.");
+                    String.format(
+                            "Currently, SnapshotBatchScanner doesn't support limit pushdown. Table: %s, bucket: %s, snapshot ID: %d, requested limit: %d",
+                            tableInfo.getTablePath(), tableBucket, snapshotId, limit));
         }
         String scannerTmpDir =
                 conn.getConfiguration().getString(ConfigOptions.CLIENT_SCANNER_IO_TMP_DIR);
@@ -123,7 +134,11 @@ public BatchScanner createBatchScanner(TableBucket tableBucket, long snapshotId)
         try {
             snapshotMeta = admin.getKvSnapshotMetadata(tableBucket, snapshotId).get();
         } catch (Exception e) {
-            throw new FlussRuntimeException("Failed to get snapshot metadata", e);
+            throw new FlussRuntimeException(
+                    String.format(
+                            "Failed to get snapshot metadata for table bucket %s, snapshot ID: %d, Table: %s",
+                            tableBucket, snapshotId, tableInfo.getTablePath()),
+                    e);
         }
 
         return new KvSnapshotBatchScanner(

fluss-client/src/main/java/org/apache/fluss/client/write/WriterClient.java

@@ -92,12 +92,16 @@ public WriterClient(
             MetadataUpdater metadataUpdater,
             ClientMetricGroup clientMetricGroup,
             Admin admin) {
+        int maxRequestSizeLocal = -1;
+        IdempotenceManager idempotenceManagerLocal = null;
         try {
             this.conf = conf;
             this.metadataUpdater = metadataUpdater;
-            this.maxRequestSize =
+            maxRequestSizeLocal =
                     (int) conf.get(ConfigOptions.CLIENT_WRITER_REQUEST_MAX_SIZE).getBytes();
-            this.idempotenceManager = buildIdempotenceManager();
+            this.maxRequestSize = maxRequestSizeLocal;
+            idempotenceManagerLocal = buildIdempotenceManager();
+            this.idempotenceManager = idempotenceManagerLocal;
             this.writerMetricGroup = new WriterMetricGroup(clientMetricGroup);
 
             short acks = configureAcks(idempotenceManager.idempotenceEnabled());
@@ -117,7 +121,14 @@ public WriterClient(
                             this::maybeAbortBatches);
         } catch (Throwable t) {
             close(Duration.ofMillis(0));
-            throw new FlussRuntimeException("Failed to construct writer", t);
+            throw new FlussRuntimeException(
+                    String.format(
+                            "Failed to construct writer. Max request size: %d bytes, Idempotence enabled: %b",
+                            maxRequestSizeLocal,
+                            idempotenceManagerLocal != null
+                                    ? idempotenceManagerLocal.idempotenceEnabled()
+                                    : false),
+                    t);
         }
     }
 
@@ -148,7 +159,11 @@ public void flush() {
         try {
             accumulator.awaitFlushCompletion();
         } catch (InterruptedException e) {
-            throw new FlussRuntimeException("Flush interrupted." + e);
+            throw new FlussRuntimeException(
+                    String.format(
+                            "Flush interrupted after %d ms. Writer may be in inconsistent state",
+                            System.currentTimeMillis() - start),
+                    e);
         }
         LOG.trace(
                 "Flushed accumulated records in writer in {} ms.",
@@ -196,7 +211,12 @@ private void doSend(WriteRecord record, WriteCallback callback) {
                 // TODO add the wakeup logic refer to Kafka.
             }
         } catch (Exception e) {
-            throw new FlussRuntimeException(e);
+            throw new FlussRuntimeException(
+                    String.format(
+                            "Failed to send record to table %s. Writer state: %s",
+                            record.getPhysicalTablePath(),
+                            sender != null && sender.isRunning() ? "running" : "closed"),
+                    e);
         }
     }
 
@@ -212,7 +232,10 @@ private void maybeAbortBatches(Throwable t) {
     private void throwIfWriterClosed() {
         if (sender == null || !sender.isRunning()) {
             throw new IllegalStateException(
-                    "Cannot perform operation after writer has been closed");
+                    String.format(
+                            "Cannot perform write operation after writer has been closed. Sender running: %b, Thread pool shutdown: %b",
+                            sender != null && sender.isRunning(),
+                            ioThreadPool == null || ioThreadPool.isShutdown()));
         }
     }
 
@@ -225,11 +248,11 @@ private IdempotenceManager buildIdempotenceManager() {
                 && maxInflightRequestPerBucket
                         > MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE) {
             throw new IllegalConfigurationException(
-                    "The value of "
-                            + ConfigOptions.CLIENT_WRITER_MAX_INFLIGHT_REQUESTS_PER_BUCKET.key()
-                            + " should be less than or equal to "
-                            + MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE
-                            + " when idempotence writer enabled to ensure message ordering.");
+                    String.format(
+                            "Invalid configuration for idempotent writer. The value of %s (%d) should be less than or equal to %d when idempotence is enabled to ensure message ordering",
+                            ConfigOptions.CLIENT_WRITER_MAX_INFLIGHT_REQUESTS_PER_BUCKET.key(),
+                            maxInflightRequestPerBucket,
+                            MAX_IN_FLIGHT_REQUESTS_PER_BUCKET_FOR_IDEMPOTENCE));
         }
 
         TabletServerGateway tabletServerGateway = metadataUpdater.newRandomTabletServerClient();
@@ -249,10 +272,9 @@ private short configureAcks(boolean idempotenceEnabled) {
 
         if (idempotenceEnabled && ack != -1) {
             throw new IllegalConfigurationException(
-                    "Must set "
-                            + ConfigOptions.CLIENT_WRITER_ACKS.key()
-                            + " to 'all' in order to use the idempotent writer. Otherwise "
-                            + "we cannot guarantee idempotence.");
+                    String.format(
+                            "Invalid acks configuration for idempotent writer. Must set %s to 'all' (current value: '%s') in order to use the idempotent writer. Otherwise we cannot guarantee idempotence",
+                            ConfigOptions.CLIENT_WRITER_ACKS.key(), acks));
         }
 
         return ack;
@@ -262,10 +284,9 @@ private int configureRetries(boolean idempotenceEnabled) {
         int retries = conf.getInt(ConfigOptions.CLIENT_WRITER_RETRIES);
         if (idempotenceEnabled && retries == 0) {
             throw new IllegalConfigurationException(
-                    "Must set "
-                            + ConfigOptions.CLIENT_WRITER_RETRIES.key()
-                            + " to non-zero when using the idempotent writer. Otherwise "
-                            + "we cannot guarantee idempotence.");
+                    String.format(
+                            "Invalid retries configuration for idempotent writer. Must set %s to non-zero (current value: %d) when using the idempotent writer. Otherwise we cannot guarantee idempotence",
+                            ConfigOptions.CLIENT_WRITER_RETRIES.key(), retries));
         }
         return retries;
     }